
//Jingle Wang 
#include "../include/replica.h"
#include "../../execution/include/exec.h"
#include <time.h>
#include <sys/time.h>

int conflict, weird, slow, happy;

int cpcounter = 0;
int64_t fzxid = 0;
//debug
int prop_cnt = 0;
bool writeToDisk = false;
bool UsingRDMA = false;

std::vector<tk_command> emptyVec_cmd;
std::vector<Propose> emptyVec_pro;
std::vector<int32_t> emptyVec_int;
std::vector<bool> emptyVec_bool;
std::unordered_map<uint64_t, uint64_t> deferMap;
std::unordered_map<int, rdma_fd *> client_connections;
std::chrono::system_clock::time_point start;
std::chrono::system_clock::time_point end;

PreAccept pa;
TryPreAccept tpa;
Prepare pr;
Accept ea;
Commit cm;
CommitShort cms;

struct timeval tv_begin, tv_end;
int64_t thistime;

Replica::Replica(){
}

Replica::Replica(int32_t _rId, std::vector<std::string> &_addrList, std::vector<int> &_portList,
				 bool _thrifty, bool _exec, bool _dreply, bool _beacon, bool _durable):
			Id(_rId), PeerAddrList(_addrList), PeerPortList(_portList), Thrifty(_thrifty),
			Exec(_exec), Dreply(_dreply), Beacon(_beacon), Durable(_durable), Debug(false) {
		}
Replica::Replica(int32_t _rId, std::vector<std::string> &_addrList, std::vector<int> &_portList,
                 bool _thrifty, bool _exec, bool _dreply, bool _beacon, bool _durable,int32_t _groupSize):
            Id(_rId), PeerAddrList(_addrList), PeerPortList(_portList), Thrifty(_thrifty),
            Exec(_exec), Dreply(_dreply), Beacon(_beacon), Durable(_durable), Debug(false), group_size(_groupSize) {
        }
bool Replica::verify() {
    if(group_size == 0) {
        group_size = GROUP_SIZE;
        //group_size = 5;
    }
    if(group_size % 2 == 0) {
        fprintf(stderr, "Group size must be odd!\n");
        return false;
    }
    if (ListeningPort < 0) {
        ListeningPort = SERVER_PORT;
    }
    if(checkpoint_cycle == 0) {
        checkpoint_cycle = CHECKPOINT_CYCLE;
    }
    if (PeerAddrList.size() == 0) {
        for (int i = 0; i < group_size; i++) {
            PeerAddrList.push_back("localhost");
            PeerPortList.push_back(SERVER_PORT + i);
        }
    }
    return true;
}


bool Replica::init() {
    Exec = true;
    group_size = 0;
    checkpoint_cycle = 0;
    //ListeningPort = -1;
//    DLOG_INIT();
    if(!verify()){
        return false;
    }
    printf("group size init to %d\n",group_size);
    if(path.empty()){
        path = "/tmp/test" + std::to_string(Id);
    }
    maxSeq = 0;
    maxSeqPerKey.clear();
    Shutdown = false;
    Restore = false;
    InstanceMatrix = new tk_instance**[group_size];
    executeUpTo.resize((unsigned long)group_size, -1);
    committedUpTo = {0};
    mq = new MsgQueue(1024 * 1024,8);
    pro_mq = new MsgQueue(1024 * 1024 * 8 ,8);
    Peers.resize((unsigned long)group_size, -1);
    Alive.resize((unsigned long)group_size, false);
    PreferredPeerOrder.resize((unsigned long)group_size, 0);
    Ewma.resize((unsigned long)group_size, 0.0);
    crtInstance.resize((unsigned long)group_size, 0);
    for(unsigned int i = 0; i < group_size; i++){
        InstanceMatrix[i] = new tk_instance *[1024 * 1024 * 8];
        memset(InstanceMatrix[i], 0, 1024 * 1024 * 8 *sizeof(tk_instance *));
        conflicts.push_back(std::unordered_map<std::string, int32_t>());
        PreferredPeerOrder[i] = int32_t(Id + 1 + i) % group_size;
    }
    if(UsingRDMA){
        rdma_ib_info = (context_info *)malloc(sizeof(context_info));
        open_device_and_alloc_pd(rdma_ib_info);
        rdma_handlers.resize((unsigned long) group_size, NULL);
    }

//    send_message_pool = new ThreadPool(1);
    if(Restore){
        //TODO: recovery from log file; (original lost)
    }
    //send_buff_queue.resize(1024 * 4, NULL);
    for(int i = 0; i < 1024 * 4; i++){
        serializeBuff *msg = create_serialize_buff();
        send_buff_queue.push(msg);
        PreAccept *pat = new PreAccept();
        preaccept_queue.push(pat);
        // printf("msg addr %p, msg buff addr:%p\n", msg, msg->buff);
    }
//    void *send_buff_queue = ;

    message_queues = (lf_queue *)malloc(sizeof(lf_queue) * (group_size + 1));

    for(int i = 0; i <= group_size; i++){
        lf_queue_init(message_queues + i, 0, 8, 1024 * 1024 * 8);
    }    

 //   lf_queue_init(&message_queue, 0, 8, 1024 * 1024 * 8);
    zServer = new zookeeperServer();
    return true;
}


bool Replica::run() { 

    init();
    std::vector<std::thread *> threads;
    connectToPeers(threads);
    threads.push_back(new std::thread(wait_for_client_connection_rdma, this));
    fprintf(stdout, "Wait for client connections .................................. DONE!\n");

    if (Exec) {
        statemachine = new Tkdatabase();
        threads.push_back(new std::thread(execute_thread, this));
        fprintf(stdout, "Start execution loop ......................................... DONE!\n");
    }

    if (Id == 0) {
        std::vector<int32_t> quorum((unsigned long)(group_size / 2 + 1), 0);
        for (int i = 0; i <= group_size / 2; i++)
            quorum[i] = i;
        updatePreferredPeerOrder(quorum);
        fprintf(stdout, "Update preferred order ....................................... DONE!\n");
    }

    // threads.push_back(new std::thread(fastClock, this));
    // fprintf(stdout, "Start fast clock ............................................. DONE!\n");

    if (Beacon) {
        threads.push_back(new std::thread(slowClock, this));
        fprintf(stdout, "Start slow clock ............................................. DONE!\n");
        threads.push_back(new std::thread(stopAdapting, this));
        fprintf(stdout, "Stop adapting ................................................ DONE!\n");
    }
    runZookeeperServer(this);
    //recovery from log file
    crtInstance = readInstanceMatrixFromFile(InstanceMatrix,group_size);
    //for(int i =0;i < crtInstance.size();i++){
    //    cout<<"crt instance"<<crtInstance.at(i)<<endl;
    //}
    for(int m = 0;m < group_size;m++){
        for(int n = 0;n < 1024 * 1024 * 8;n++){
            if(InstanceMatrix[m][n] != nullptr && InstanceMatrix[m][n]->status != COMMITTED)
                startRecoveryForInstance(m,n);
        }
    }

    int i;
    MsgQueue * pro_mq_s = pro_mq;
    fprintf(stdout, "Start message dispatcher ..................................... DONE!\n");
    void * msgp = malloc(8);
    Prepare *prepare = nullptr;
    PreAccept *preAccept = nullptr;
    Accept *accept = nullptr;
    Commit *commit = nullptr;
    CommitShort *commitshort = nullptr;
    PrepareReply *prepare_reply = nullptr;
    PreAcceptOk *preAcceptOk = nullptr;
    PreAcceptReply *preAcceptReply = nullptr;
    AcceptReply *acceptReply = nullptr;
    TryPreAccept *tryPreAccept = nullptr;
    Propose * msgpro = nullptr;
    lf_queue message_queue = nullptr;
    while (!Shutdown) {
        
        int queue_id = find_longest_queue(message_queues, group_size + 1);

        if(queue_id < 0){
            continue;
        }
       // cout<<"queue_id = "<<queue_id<<endl;

//        int ret = lf_queue_pop(mq, msgp);
//        if(ret < 0){
//
//        }
        message_queue = *(message_queues + queue_id);
        lf_queue_pop(message_queue, msgp);
        //msgp = mq->get();
        // if(msgp == nullptr){
        //     continue;
        // }
        // if(msgp == nullptr){
        //     continue;
        // }
        if(*(TYPE **)msgp == nullptr){
            continue;
        }

        TYPE msgType = **(TYPE **) msgp;
        // if (Debug && msgType != FAST_CLOCK) {
        // }
       // cout<<"replica process msgType"<<msgType<<endl;
        switch (msgType) {
            case FAST_CLOCK:
                pro_mq_s = pro_mq;
                break;
            case PROPOSE:
                msgpro = *(Propose **) msgp;
                handlePropose(msgpro);
                break;
            case PREPARE:
                // TODO: LOG
//                if (Debug)
//                    fprintf(stdout, "going to handle prepare\n");
                prepare = *(Prepare **) msgp;
                if(prepare == nullptr){
             //       DLOG_LOG(ERROR,"*(Prepare **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
             //   DLOG_LOG(DEBUG,"Process Prepare msg, current mq size: %d, Command Leader: %d, Instance: %d",
              //           mq->count(), prepare->LeaderId, prepare->Instance);
                handlePrepare(prepare);
                delete prepare;
                prepare = nullptr;

                break;
            case PREACCEPT:
                // TODO: LOG
//                if (Debug)
//                    fprintf(stdout, "going to handle preaccept\n");
                preAccept =  *(PreAccept **) msgp;
                if(preAccept == nullptr){
                    break;
                }

                handlePreAccept(preAccept);
                delete preAccept;
                preAccept = nullptr;
                break;
            case ACCEPT:
//                if (Debug)
//                    fprintf(stdout, "going to handle accept\n");
                accept = *(Accept **) msgp;
                if(accept == nullptr){
                    break;
                }
             //   DLOG_LOG(DEBUG,"Process Accept msg, current mq size: %d, Command Leader: %d, Instance: %d",
            //             mq->count(), accept->LeaderId, accept->Instance);
                handleAccept(accept);
                delete accept;
                accept = nullptr;
                break;
            case COMMIT:
//                if (Debug)
//                    fprintf(stdout, "going to handle commit\n");
                commit = *(Commit **)msgp;
                if(commit == nullptr){
             //       DLOG_LOG(ERROR,"*(Commit **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
              //  DLOG_LOG(DEBUG,"Process Commit msg, current mq size: %d, Command Leader: %d, Instance: %d",
              //           mq->count(), commit->LeaderId, commit->Instance);
                handleCommit(commit);
                delete commit;
                commit = nullptr;
                break;
            case COMMIT_SHORT:
//                if (Debug)
//                    fprintf(stdout, "going to handle commit short\n");
                commitshort = *(CommitShort **)msgp;
                if(commitshort == nullptr){
              //      DLOG_LOG(ERROR,"*(CommitShort **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
             //   DLOG_LOG(DEBUG, "Process CommitShort msg, current mq size: %d, Command Leader: %d, Instance: %d",
              //           mq->count(), commitshort->LeaderId, commitshort->Instance);
                handleCommitShort(commitshort);
                delete commitshort;
                commitshort = nullptr;
                break;
            case PREPARE_REPLY:
//                if (Debug)
//                    fprintf(stdout, "going to handle prepare reply\n");
                prepare_reply = *(PrepareReply **)msgp;
                if(prepare_reply == nullptr){
             //       DLOG_LOG(ERROR,"*(PrepareReply **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
             //   DLOG_LOG(DEBUG,"Process PrepareReply msg, current mq size: %d, Command Leader: %d, Instance: %d",
              //           mq->count(), prepare_reply->Replica, prepare_reply->Instance);
                handlePrepareReply(prepare_reply);
                delete prepare_reply;
                prepare_reply = nullptr;
                break;
            case PREACCEPT_REPLY:
//                if (Debug)
//                    fprintf(stdout, "going to handle preaccept reply\n");
                preAcceptReply = *(PreAcceptReply **)msgp;
                if(preAcceptReply == nullptr){
              //      DLOG_LOG(ERROR,"*(PreAcceptReply **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
              //  DLOG_LOG(DEBUG,"Process PreAcceptReply msg, current mq size: %d, Command Leader: %d, Instance: %d",
              //           mq->count(), preAcceptReply->Replica, preAcceptReply->Instance);
                handlePreAcceptReply(preAcceptReply);
                delete preAcceptReply;
                preAcceptReply = nullptr;
                break;
            case PREACCEPT_OK:
//                if (Debug)
//                    fprintf(stdout, "going to handle preaccept ok\n");
                //gettimeofday(&tv_end, NULL);
                //thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
                //cout<<"the latency when I get a preacceptOK"<<thistime<<endl;
                preAcceptOk = *(PreAcceptOk **)msgp;
                if(preAcceptOk == nullptr){
               //     DLOG_LOG(ERROR,"*(PreAcceptOk **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
               // DLOG_LOG(DEBUG,"Process PreAcceptOk msg, current mq size: %d, Instance: %d",
              //           mq->count(),preAcceptOk->Instance);
                handlePreAcceptOK(preAcceptOk);
                delete preAcceptOk;
                preAcceptOk = nullptr;
                break;
            case ACCEPT_REPLY:
//                if (Debug)
//                    fprintf(stdout, "going to handle accept reply\n");
                acceptReply = *(AcceptReply **)msgp;
                if(acceptReply == nullptr){
               //     DLOG_LOG(ERROR,"*(AcceptReply **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
              //  DLOG_LOG(DEBUG,"Process AcceptReply msg, current mq size: %d, Sender: %d, Instance: %d",
              //           mq->count(), acceptReply->Replica, acceptReply->Instance);
                handleAcceptReply(acceptReply);
                delete acceptReply;
                acceptReply = nullptr;
                break;
            case TRY_PREACCEPT:
//                if (Debug)
//                    fprintf(stdout, "going to handle try preaccept\n");

                if(*(TryPreAccept **) msgp == nullptr){
                //    DLOG_LOG(ERROR,"*(TryPreAccept **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
               // DLOG_LOG(DEBUG,"Process TryPreAccept msg, current mq size: %d", mq->count());
                handleTryPreAccept(*(TryPreAccept **) msgp);
                break;
            case TRY_PREACCEPT_REPLY:
//                if (Debug)
//                    fprintf(stdout, "going to handle try preaccept reply\n");
                if(*(TryPreAcceptReply **) msgp == nullptr){
                 //   DLOG_LOG(ERROR,"*(TryPreAcceptReply **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
               // DLOG_LOG(DEBUG,"Process TryPreAcceptReply msg, current mq size: %d", mq->count());
                handleTryPreAcceptReply(*(TryPreAcceptReply **) msgp);
                break;
            case BEACON:
//                fprintf(stdout, "going to handle beacon\n");
                replyBeacon(*(Beacon_msg **) msgp);
                break;
            case SLOW_CLOCK:
//                if (Debug)
//                    fprintf(stdout, "going to handle slow clock\n");
                // if (Beacon) {
                //     for (i = 0; i < group_size; i++) {
                //         if (i == Id)
                //             continue;
                //         sendBeacon(i);
                //     }
                // }
                break;
            case CLIENT_CONNECT:
//                if (Debug)
//                    fprintf(stdout, "going to handle client connect\n");
                weird = conflict = slow = happy = 0;
                break;
            case RECOVER_INSTANCE:
//                if (Debug)
//                    fprintf(stdout, "going to handle recovery instance\n");
                if(*(InstanceId **) msgp == nullptr){
                   // DLOG_LOG(ERROR,"*(InstanceId **)msgp is nullptr, current mq size: %d", mq->count());
                    break;
                }
                //DLOG_LOG(DEBUG,"Process InstanceId msg, current mq size: %d", mq->count());
                startRecoveryForInstance((*(InstanceId **) msgp)->replica, (*(InstanceId **) msgp)->instance);
                break;
            default:
                break;
        }
        //msgp = nullptr;
    }
    free(msgp);

    //DONE:  1. event loop;    -- To process the message from servers each other, get msgs from a channel;
    //DONE:  2. excution Loop; -- periodically to apply cmds. build the KV;
    //DONE:  3. propose Loop;  -- client raise a proposal, when server receives a cmd, process it, and put it in a Propose channel;
    //DONE:  4. Timeout Loop;  -- check timeout,
    //DONE:  5. network start; -- connect peer and peer, and process the message.
    //TODO:  6. network communication
    for (i = 0; i < threads.size(); i++) {
        if (threads[i]->joinable())
            threads[i]->detach();
        delete threads[i];
    }
    return true;
}


void runZookeeperServer(Replica * r){
    bool initDone = initZookeeperServer(r->zServer,r->ListeningPort + 100,r->Id);
    cout<<"init zookeeperServer done, starting zookeeperServer thread"<<endl;
    thread acceptFromClients(listenOnPort,r->zServer,r->message_queues+r->group_size);
    thread pre_process(pre_processPacket,r->zServer,r->message_queues+r->group_size);
    thread process(processPacket,r->zServer);
    thread sendResponseToClients(sendResponse,r->zServer);
    thread sessionTracker(processSessionTracker,&r->zServer->cnxnFactory.st);

    acceptFromClients.detach();
    pre_process.detach();
    process.detach();
    sendResponseToClients.detach();
    sessionTracker.detach();
}
void wait_for_client_connection_rdma(Replica *r){
    ClientConnect * notice = new ClientConnect();
    std::vector<std::thread *> clientListeners;
    while (!r->Shutdown) {
        // TODO - DONE
        // accept requests from clients
        // RDMA_CONNECTION conn = r->Listener->accept();
        // _ClientParam * cpr = new _ClientParam(r, conn);
        // go(clientListener, cpr);
        int sock = acceptAt(r->Listener);
        rdma_fd *handler = (rdma_fd *)malloc(sizeof(rdma_fd));
        handler->fd = sock;
        get_context_info(handler, r->rdma_ib_info);
        build_rdma_connection(handler);
        printf("connection complete!\n");
//        query_qp(handler);
        client_connections[sock] = handler;
        clientListeners.push_back(new std::thread(client_listener_rdma, r, handler));
        //if (r->Debug) {
        fprintf(stdout, "New client connected.\n");
        //}
        lf_queue_push(*(r->message_queues + r->group_size), &notice);
    //    r->mq->put(&notice);
    }
    for (std::thread * cl: clientListeners) {
        cl->detach();
        delete cl;
    }
    delete notice;
}
void client_listener_rdma(Replica * r, rdma_fd* handler) {
    uint8_t msg_type = -1;
//    Read read;
//    ProposeAndRead pandr;
    // debug
//    int32_t prop_count = 0;
    while (!r->Shutdown) {
        // TODO - DONE
        // msgType = reader->ReadByte();
        // if ERROR: return;
        // TEST
        //serializeBuff *message = new serializeBuff();
        serializeBuff * message ;
        //message->len = 0;
        //message->buff = NULL;
        if(handler->have_read + sizeof(uint8_t) + sizeof(uint32_t) + 1 > handler->receive_buf_size){
            handler->have_read = 0;
        }
        while(1){
            volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + 5);
            if(*end == '1'){
                break;
            }
        }
        memcpy(&msg_type, handler->receive_buf + handler->have_read, sizeof(uint8_t));
        int32_t len = 0;
        memcpy(&len, handler->receive_buf + handler-> have_read + sizeof(uint8_t), sizeof(uint32_t));
        //msg_type = *((uint8_t *)handler->receive_buf + handler->have_read);
        //int32_t len = *((uint32_t *)handler->receive_buf + handler->have_read + sizeof(uint8_t));
        //printf("1.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
        memset(handler->receive_buf + handler->have_read, 0, sizeof(uint8_t) + sizeof(uint32_t) + 1);
        handler->have_read += sizeof(uint8_t) + sizeof(uint32_t) + 1;
        if(handler->have_read + len + 1 > handler->receive_buf_size){
            handler->have_read = 0;
        }
        while(1){
            volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + len);
            if(*end == '1'){
                break;  
            }
        }
        //printf("2.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
        char * buff = (char *)malloc(len);
        memcpy(buff, handler->receive_buf + handler->have_read, len);
        message = create_serialize_buff(buff,len);
        memset(handler->receive_buf + handler->have_read, 0, len + 1);
        handler->have_read += len + 1;
//        printf("before dispatch, message: %p, message->len: %d\n", message, message->len);

 //       r->msg_dispatcher(remote_id, message, (TYPE)msg_type);
        
        int ret = 0;
        if((TYPE)msg_type == PROPOSE){
            Propose * prop = new Propose();
            ret = deserializePropose(message, prop);
            prop->Conn = handler->fd;
            // ProposeReplyTS * prts = new ProposeReplyTS(true, prop->CommandId, 0,
            //                                                "", 0);
            // int ret = reply_propose_ts_rdma(prts, handler);
            // if(ret == -1){
            //         printf("reply to client failed\n");
            // }
            // delete prts;
            // prts = nullptr;
            //r->pro_mq->put(&prop);
            lf_queue_push(*(r->message_queues + r->group_size), &prop);
            //lf_queue_push(r->message_queue, &prop);
//            r->mq->put(&prop);
        }
        delete message;
    }
}

void replicaListener(Replica * r, int32_t rid, RDMA_CONNECTION conn) {
    uint8_t msgType;
    Beacon_msg gbeacon;
    Beacon_msg_reply gbeaconReply;
    Beacon_msg * beacon;
    int len = 0;

    while (!r->Shutdown) {
        // TODO - DONE
        // msgType = reader->ReadByte();
        // if ERROR: return;
        // TEST
        //fprintf(stdout, "Ready to read msgType from replica %d\n", rid);
        //fflush(stdout);
        if (readUntil(conn, (char *)&msgType, 1) < 0) {
            r->Alive[rid] = false;
            fprintf(stderr, "Connection between replica %d disconnect!\n", rid);
            destroyConnection(conn);
            return;
        }
        switch ((TYPE)(msgType)) {
            case BEACON:
                if ( !gbeacon.Unmarshal(conn) )
                    return;
                beacon = new Beacon_msg(rid, gbeacon.timestamp);
                r->mq->put(&beacon);
                break;
            case BEACON_REPLY:
                if ( !gbeaconReply.Unmarshal(conn) )
                    return;
                r->Ewma[rid] = 0.99*r->Ewma[rid] + 0.01*(double)(CPUTicks() - gbeaconReply.timestamp);
                break;
            default:
                // TODO - DONE:
                // find matched msg Type
                // if find: unmarshal new msg
                // else: LOG("Unknown msg type");
                // putIntoMsgQueue(r->mq, msg);
                // TEST
                if((TYPE)msgType >= 33){
                    //DLOG_LOG(DEBUG, "Replica %d have recieve unknown msgType from Replica %d", r->Id,  rid);
                }
                readUntil(conn,(char *)&len, 4);
                char * buff = (char *)malloc(len);
                readUntil(conn,buff,len);
                serializeBuff * message = create_serialize_buff(buff,len);
		        int32_t ret = r->msg_dispatcher(rid, message, (TYPE)msgType);
                //r->msgDispatcher(rid, conn, (TYPE)msgType);
                break;
        }
    }
}

void waitForPeerConnections(Replica * r, bool * done) {
    // TODO - DONE
    // r->Listener = RDMA_Listen("tcp", r->peerAddrList[Id]);
    // TEST
    do {
        r->Listener = listenOn(r->ListeningPort);
        if (r->Listener < 0) {
            fprintf(stderr, "Cannot listen on 0.0.0.0:%d\n", r->ListeningPort);
            std::this_thread::sleep_for(std::chrono::seconds(3));
        }
    } while (r->Listener < 0);
    for (int i = r->Id + 1; i < r->group_size; i++) {
        // TODO - DONE
        int conn = acceptAt(r->Listener);
        if (conn < 0) {
            fprintf(stderr, "Error when accept connect request from other replicas\n");
            continue;
        }
        int32_t id;
        if (readUntil(conn, (char *) &id, 4) < 0) {
            fprintf(stderr, "Error reading replica id!\n");
            destroyConnection(conn);
            i--;
            continue;
        }
        fprintf(stdout, "Connect with replica %d ....................................... DONE!\n", id);
        r->Peers[id] = conn;
        if(UsingRDMA){
            rdma_fd *handler = (rdma_fd *)malloc(sizeof(rdma_fd));
            handler->fd = conn;
            get_context_info(handler, r->rdma_ib_info);
            build_rdma_connection(handler);
            printf("connection complete!\n");
            query_qp(handler);
            r->rdma_handlers[id] = handler;
        }
        
        // r->PeerReaders[id] = NewReader(conn)
        // r->PeerWriters[id] = NewWriter(conn)
        r->Alive[id] = true;
    }
    *done = true;
}
void Replica::connectToPeers(std::vector<std::thread *> & threads) {

    bool done = false;
    std::thread wfpr(waitForPeerConnections, this, &done);
    wfpr.detach();
    //connect to peers
    for (int i = 0; i < Id; i++) {
        bool finish = false;
        while (!finish) {
            // TODO - DONE
            // RDMA_CONNECTION conn == Dial(PeerAddrList[i]);
            // if (conn != ERROR) {
            //     Peers[i] = conn;
            //     finish = true;
            // } else
            //     nano_sleep(1000 * 1000 * 1000);
            // TEST
            int conn = dialTo(PeerAddrList[i], (uint16_t)PeerPortList[i]);
            if (conn < 0) {
                fprintf(stdout, "Cannot dial to replica %d(%s:%d)\n", i, PeerAddrList[i].c_str(), PeerPortList[i]);
                std::this_thread::sleep_for(std::chrono::seconds(1));
            } else {
                Peers[i] = conn;
                finish = true;
            }
        }
        // TODO - DONE
        // if (RDMA_Write(Peers[i], b) == ERROR) {
        //    LOG("Write Id error!");
        //    continue;
        // }
        // TEST
        if(UsingRDMA){
            int32_t tmp = Id;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            if (sendData(Peers[i], (char *) &tmp, 4) != 4) {
                printf("Write Id error!\n");
                continue;
            }
            rdma_fd *handler = (rdma_fd *)malloc(sizeof(rdma_fd));
            handler->fd = Peers[i];               
            get_context_info(handler, rdma_ib_info);
            build_rdma_connection(handler);
            rdma_handlers[i] = handler;
            Alive[i] = true;
            query_qp(handler);
        }else{
            int32_t tmp = Id;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            if (sendData(Peers[i], (char *) &tmp, 4) != 4) {
                fprintf(stderr, "Write Id error!\n");
                continue;
            }
            Alive[i] = true;
        }
        
        // PeerReaders[i] = NewReader(Peers[i])
        // PeerWriters[i] = NewWriter(Peers[i])
    }

    while (!done);
//    TODO - DONE
//    LOG("Replica id: %d. Done connecting to peers\n", Id);
//    for (int32_t rid = 0; rid < group_size; rid++) {
//        if (rid == Id)
//            continue;
//        _ListenerParam * lpr = new _LIstenerParam(this, rid, reader);
//        go(replicaListener, lpr);
//    }
    fprintf(stdout, "Replica %d, connect to peers .................................. DONE!\n", Id);
    for (int32_t rid = 0; rid < group_size; rid++) {
        if (rid == Id)
            continue;
        if(UsingRDMA){
            threads.push_back(new std::thread(rdma_replica_listener, this, rid, rdma_handlers[rid]));
        }else{
            threads.push_back(new std::thread(replicaListener, this, rid, Peers[rid]));
        }
        
        threads.back()->detach();
        fprintf(stdout, "Create listener for replica %d ................................ DONE!\n", rid);
    }
}



serializeBuff* Replica::serialize_message(TYPE msg_type, void *message){
	//serializeBuff *oa = new serializeBuff();
    serializeBuff* oa = nullptr;
    if(send_buff_queue.empty()){
        for(int i = 0; i < 1024; i++){
            oa = create_serialize_buff();
            send_buff_queue.push(oa);
        }

    }
    oa =send_buff_queue.front();
    send_buff_queue.pop();
    oa->off = 0;
    
    int ret = 0;
	switch(msg_type){
		case BEACON:
			serializeBeacon_msg((Beacon_msg *)message, oa);
            break;
		//TODO: other message;
        case BEACON_REPLY:
            serializeBeacon_msg_reply((Beacon_msg_reply *)message, oa);
        case BE_LEADER:
            break;
        case GET_LEADER_ARGS:
            serializeGetLeaderArgs((GetLeaderArgs *)message, oa);
            break;
        case GET_LEADER_REPLY:
            serializeGetLeaderReply((GetLeaderReply *)message, oa);
            break;
        case GET_REPLICA_LIST_ARGS:
            serializeGetReplicaListArgs((GetReplicaListArgs *)message, oa);
            break;
        case GET_REPLICA_LIST_REPLY:
            serializeGetReplicaListReply((GetReplicaListReply *)message, oa);
            break;
        case REGISTER_ARGS:
            serializeRegisterArgs((RegisterArgs *)message, oa);
            break;
        case REGISTER_REPLY:
            serializeRegisterReply((RegisterReply *)message, oa);
            break;
        case INSTANCE_ID:
            serializeInstanceId((InstanceId *)message, oa);
            break;
        case CLIENT_CONNECT:
            break;
        case FAST_CLOCK:
            break;
        case SLOW_CLOCK:
            break;
        case TRY_PREACCEPT:
            serializeTryPreAccept((TryPreAccept *)message, oa);
            break;
        case TRY_PREACCEPT_REPLY:
            serializeTryPreAcceptReply((TryPreAcceptReply *)message, oa);
            break;
        case ACCEPT_REPLY:
            serializeAcceptReply((AcceptReply *)message, oa);
            break;
        case PREACCEPT_OK:
            serializePreAcceptOk((PreAcceptOk *)message, oa);
            break;
        case PREACCEPT_REPLY:
            serializePreAcceptReply((PreAcceptReply *)message, oa);
            break;
        case PREPARE_REPLY:
            serializePrepareReply((PrepareReply *)message, oa);
            break;
        case COMMIT_SHORT:
            serializeCommitShort((CommitShort *)message, oa);
            break;
        case COMMIT:
            serializeCommit((Commit *)message, oa);
            break;
        case ACCEPT:
            serializeAccept((Accept *)message, oa);
            break;
        case PREACCEPT:
            ret = serializePreAccept((PreAccept *)message, oa);
            break;
        case PREPARE:
            serializePrepare((Prepare *)message, oa);
            break;
        case PROPOSE_REPLY_TS:
            serializeProposeReplyTS((ProposeReplyTS *)message, oa);
            break;
        case PROPOSE_REPLY:
            serializeProposeReply((ProposeReply *)message, oa);
            break;
        case PROPOSE_AND_READ:
            serializeProposeAndRead((ProposeAndRead *)message, oa);
            break;
        case PROPOSE:
            serializePropose((Propose *) message, oa);
            break;
        case READ:
            serializeRead((Read *)message, oa);
            break;
	}
	return oa;
}


int send_message(rdma_fd *handler, serializeBuff *message, TYPE msg_type){
    int ret = 0;
    int send_buf_size = sizeof(uint8_t) + sizeof(uint32_t) + 1 + message->off + 1;
    char *send_buf = (char *)malloc(send_buf_size);
    gen_send_buf(send_buf, message, msg_type);
    if(handler != 0)
        ret = rdma_write(handler, send_buf, send_buf_size);
    free(send_buf);
    return ret;
}

int send_message(int sock,serializeBuff * message,TYPE msg_type){
    //msgtype
    uint8_t msgType = (uint8_t) msg_type;
    sendData(sock, (char *) & msgType, 1);
    //len
    sendData(sock,(char *)&message->off,4);
    //data
    sendData(sock,message->buff,message->off);
    return 0;
}


void rdma_replica_listener(Replica *r, int32_t remote_id, rdma_fd *handler){
	uint8_t msg_type = 0;
	while(!r->Shutdown){
        //serializeBuff *message = new serializeBuff();
        //message->len = 0;
        //message->buff = NULL;
        int32_t len = 0;
        char * buff;
		if(handler->have_read + sizeof(uint8_t) + sizeof(uint32_t) + 1 > handler->receive_buf_size){
			handler->have_read = 0;
		}
		while(1){
			volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + 5);
			if(*end == '1'){
				break;
			}
		}
        memcpy(&msg_type, handler->receive_buf + handler->have_read, sizeof(uint8_t));
        memcpy(&len, handler->receive_buf + handler-> have_read + sizeof(uint8_t), sizeof(uint32_t));
		//msg_type = *((uint8_t *)handler->receive_buf + handler->have_read);
		//message->len = *((uint32_t *)handler->receive_buf + handler->have_read + sizeof(uint8_t));
        //printf("1.have receive msg from id: %d, msg off: %d\n", remote_id, len);
		memset(handler->receive_buf + handler->have_read, 0, sizeof(uint8_t) + sizeof(uint32_t) + 1);
		handler->have_read += sizeof(uint8_t) + sizeof(uint32_t) + 1;
		if(handler->have_read + len + 1 > handler->receive_buf_size){
			handler->have_read = 0;
		}
		while(1){
			volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + len);
			if(*end == '1'){
				break;  
			}
		}
        //printf("2.have receive msg from id: %d, msg off: %d\n", remote_id, len);
		buff = (char *)malloc(len);
		memcpy(buff, handler->receive_buf + handler->have_read, len);
		memset(handler->receive_buf + handler->have_read, 0, len + 1);
		handler->have_read += len + 1;
//        printf("before dispatch, message: %p, message->len: %d\n", message, message->len);
        serializeBuff * message = create_serialize_buff(buff,len);
		int32_t ret = r->msg_dispatcher(remote_id, message, (TYPE)msg_type);
        if(ret > 0){
            message->off = 0;
            r->msg_dispatcher(remote_id, message, (TYPE)ret);
        }
        close_serialize_buff(message);
	}
}
/*
 *   Message dispatch
 *
 */
int32_t Replica::msg_dispatcher(uint32_t remote_id, serializeBuff *msg, TYPE msg_type){
	int ret = 0;
    lf_queue message_queue = *(message_queues + remote_id);
    //printf("msg len=%d,msg off=%d\n",msg->len,msg->off);
    if(msg_type == BEACON){
		Beacon_msg *beacon = new Beacon_msg();
		ret = deserializeBeacon_msg(msg, beacon);
	     // free the buff of msg;
        if(ret == -1){
            delete beacon;
            return -1;
        }
        lf_queue_push(message_queue, &beacon);
		//mq->put(&beacon);
        return 0;
	}
	else if(msg_type == BEACON_REPLY){
		Beacon_msg_reply *gbeacon_reply = new Beacon_msg_reply();
		ret = deserializeBeacon_msg_reply(msg, gbeacon_reply);
        if(ret == -1){
            delete gbeacon_reply;
            return -1;
        }
		Ewma[remote_id] = 0.99 * Ewma[remote_id] + 0.01 * (double)(CPUTicks() - gbeacon_reply->timestamp);
		//delete msg;
		delete gbeacon_reply;
        return 0;
	}
	else if(msg_type == PREACCEPT){
		PreAccept *pre_accept = new PreAccept();
		ret = deserializePreAccept(msg, pre_accept);
		//delete msg;
        if(ret == -1){
            delete pre_accept;
            return -1;
        }
        lf_queue_push(message_queue, &pre_accept);
		//mq->put(&pre_accept);
        return 0;
	}
	else if(msg_type == PREACCEPT_OK){
		PreAcceptOk *pre_accept_ok = new PreAcceptOk();
		ret = deserializePreAcceptOk(msg, pre_accept_ok);
		//delete msg;
        if(ret == -1){
            delete pre_accept_ok;
            return -1;
        }
        lf_queue_push(message_queue, &pre_accept_ok);
		//mq->put(&pre_accept_ok);
        return 0;
	}
	else if(msg_type == PREACCEPT_REPLY){
		PreAcceptReply *pre_accept_reply = new PreAcceptReply();
		ret = deserializePreAcceptReply(msg, pre_accept_reply);
		//delete msg;
        if(ret == -1){
            delete pre_accept_reply;
            return -1;
        }
        lf_queue_push(message_queue, &pre_accept_reply);
		//mq->put(&pre_accept_reply);
        return 0;
	}
	else if(msg_type == PREPARE){
		Prepare *prepare = new Prepare();
		ret = deserializePrepare(msg, prepare);
		//delete msg;
        if(ret == -1){
            delete prepare;
            return -1;
        }
        lf_queue_push(message_queue, &prepare);
		//mq->put(&prepare);
        return 0;
	}
	else if(msg_type == PREPARE_REPLY){
		PrepareReply *prepare_reply = new PrepareReply();
		ret = deserializePrepareReply(msg, prepare_reply);
		//delete msg;
        if(ret == -1){
            delete prepare_reply;
            return -1;
        }
        lf_queue_push(message_queue, &prepare_reply);
		//mq->put(&prepare_reply);
        return 0;
	}
	else if(msg_type == ACCEPT){
		Accept *accept = new Accept();
		ret = deserializeAccept(msg, accept);
        if(ret == -1){
            delete accept;
            return -1;
        }

		//delete msg;
        lf_queue_push(message_queue, &accept);
		//mq->put(&accept);
        return 0;
	}
	else if(msg_type == ACCEPT_REPLY){
		AcceptReply *accept_reply = new AcceptReply();
		ret = deserializeAcceptReply(msg, accept_reply);
		//delete msg;
        if(ret == -1){
            delete accept_reply;
            return -1;
        }
        lf_queue_push(message_queue, &accept_reply);        
		//mq->put(&accept_reply); 
        return 0;
	}
	else if(msg_type == COMMIT){
		Commit *commit = new Commit();
		ret = deserializeCommit(msg, commit);
        //commit->print();
        if(ret == -1){
            delete commit;
            return -1;
        }

		//delete msg;
        lf_queue_push(message_queue, &commit);
		//mq->put(&commit);	
        return 0;
	}
	else if(msg_type == COMMIT_SHORT){
		CommitShort *commit_short = new CommitShort();
		ret = deserializeCommitShort(msg, commit_short);
        /*cout<<"commit short type="<<commit_short->Type<<endl;
        if(commit_short->Type != COMMIT_SHORT){
            return (int32_t)commit_short->Type;
        }*/
		//delete msg;
        //commit_short->print();
        if(ret == -1){
            delete commit_short;
            return -1;
        }
        lf_queue_push(message_queue, &commit_short);    
		//mq->put(&commit_short);
        return 0;
	}
	else if(msg_type == TRY_PREACCEPT){
		TryPreAccept *try_pre_accept = new TryPreAccept();
		ret = deserializeTryPreAccept(msg, try_pre_accept);
		//delete msg;
        if(ret == -1){
            delete try_pre_accept;
            return -1;
        }
        lf_queue_push(message_queue, &try_pre_accept); 
		//mq->put(&try_pre_accept);
        return 0;
	}
	else if(msg_type == TRY_PREACCEPT_REPLY){
		TryPreAcceptReply *try_pre_accept_reply = new TryPreAcceptReply();
		ret = deserializeTryPreAcceptReply(msg, try_pre_accept_reply);
		//delete msg;
        if(ret == -1){
            delete try_pre_accept_reply;
            return -1;
        }
        lf_queue_push(message_queue, &try_pre_accept_reply);
		//mq->put(&try_pre_accept_reply);
        return 0;
	}
}


void slowClock(Replica * r) {
    Clock * timeout = new Clock(SLOW_CLOCK);
    while (!r->Shutdown) {
        std::this_thread::sleep_for(std::chrono::milliseconds(150)); // 150 ms
        //lf_queue_push(r->message_queue, &timeout);
        //r->mq->put(&timeout);
    }
    delete timeout;
    timeout = nullptr;
}

void fastClock(Replica * r) {
    Clock * timeout = new Clock(FAST_CLOCK);
    while (!r->Shutdown) {
        std::this_thread::sleep_for(std::chrono::microseconds(4)); // 10us
        //std::this_thread::sleep_for(std::chrono::milliseconds(5));
        //lf_queue_push(r->message_queue, &timeout);
        //r->mq->put(&timeout);
    }
    delete timeout;
    timeout = nullptr;
}

void stopAdapting(Replica * r) {
    std::this_thread::sleep_for(std::chrono::seconds(ADAPT_TIME_SEC));
    r->Beacon = false;
    std::this_thread::sleep_for(std::chrono::seconds(1));

    for (int i = 0; i < r->group_size - 1; i++) {
        int min = i;
        for (int j = i+1; j < r->group_size - 1; j++) {
            if (r->Ewma[r->PreferredPeerOrder[j]] < r->Ewma[r->PreferredPeerOrder[min]])
                min = j;
        }
        int32_t temp = r->PreferredPeerOrder[i];
        r->PreferredPeerOrder[i] = r->PreferredPeerOrder[min];
        r->PreferredPeerOrder[min] = temp;
    }
}


/****************************************************************************************
 *                                     PHASE 1                                          *
****************************************************************************************/
void Replica::handlePropose(Propose * propose) {
    //long batchSize = pro_mq->count() + 1;
    lf_queue message_queue = *(message_queues + group_size);
    long batchSize = lf_queue_len(message_queue) + 1;
    //printf("handler propose begin\n");
    if (batchSize <= 0)
        return;
    if (batchSize > MAX_BATCH) {
        batchSize = MAX_BATCH;
    }
    if (Debug) {
        //fprintf(stdout, "handle proposal %d to %ld\n", prop_cnt, prop_cnt + batchSize);
        prop_cnt += batchSize;
    }

    int32_t instNo = crtInstance[Id];
    crtInstance[Id]++;
    // TODO
    // LOG: starting instance [instNo]
    // LOG: batching [batchSize]

    std::vector<tk_command> cmds((unsigned long)batchSize, tk_command());
    std::vector<Propose> proposals((unsigned long)batchSize, Propose());
    cmds[0] = propose->Command;
    proposals[0] = *propose;
    delete propose;
    propose = nullptr;
    // the original messages can be freed since they are already copied
    // allocate a new block of space for cmds/proposals for using cache
    // since the original messages may be separated in many distinct blocks
    void *pro = malloc(8);
    for (int i = 1; i < batchSize; i++) {
        lf_queue_pop(message_queue, pro);
        propose = *(Propose **)pro;
        cmds[i] = propose->Command;
        proposals[i] = *propose;
        delete propose;
        propose = nullptr;

    }
    free(pro);
    // end = std::chrono::system_clock::now();
    // std::chrono::duration<double> diff = end-start;
    // printf("[TIME From recv to handerpropose]: %lf us\n", (double) diff.count() * 1000000);
    //printf("handler propose end\n");
    startPhase1(Id, instNo, 0, proposals, cmds, batchSize);
}

void Replica::startPhase1(int32_t replica, int32_t instance, int32_t ballot,
                          std::vector<Propose> & proposals,
                          std::vector<tk_command> & cmds, long batchSize) {
    // init command attributes
    int32_t  seq = 0;
    std::array<int32_t, GROUP_SIZE> deps;
    for (int q = 0; q < group_size; q++) {
        deps[q] = -1;
    }
    //deps[Id] = instance - 1;
    //cout<<"start phase1 update attributes for instance "<<instance<<endl;
    updateAttributes(cmds, seq, deps, Id, instance);
    //start get time
    gettimeofday(&tv_begin, NULL);
    std::vector<int32_t> _cmDeps(GROUP_SIZE, -1);
    lb_t * lb = new lb_t(proposals, 0, 0, true, 0, 0, 0, deps, _cmDeps, nullptr, false, false, emptyVec_bool, 0);
    if (lb == nullptr) {
        fprintf(stderr, "Cannot alloc new space for lb in phase 1...\n");
    }
    tk_instance * newInstance = new tk_instance(cmds, ballot, PREACCEPTED, seq, deps, lb);
    if (Debug)
        fprintf(stdout, "create new instance in matrix[%d][%d]\n", Id, instance);
    InstanceMatrix[Id][instance] = newInstance;

    updateConflicts(cmds, Id, instance, seq);

    if (seq >= maxSeq) {
        maxSeq = seq + 1;
    }

    /* TODO
     * record into stable storage
     * recordInstanceMetaData(InstanceMatrix[Id][instance]);
     * recordCommands(cmds);
     * sync(); // sync with stable storage
     */
    if(writeToDisk){
        fzxid = Id;
        fzxid = fzxid << 32;
        fzxid += instance;
        cout<<"write instance "<<fzxid<<"to file"<<endl;
        writeInstanceToFile(newInstance,fzxid);
    }
    
    //sync
    bcastPreAccept(Id, instance, ballot, cmds, seq, deps);
    cpcounter += batchSize;

    if (Id == 0 && DO_CHECKPOINTING && cpcounter >= CHECKPOINT_PERIOD) {
        // by default this block will not execute
        /*
         * cpcounter = 0;
         * crtInstance[Id]++;
         * instance++;
         * maxSeq++;
         * ....  ignored here
         */
    }
}

void Replica:: handlePreAccept(PreAccept * preAccept) {
    if(preAccept->Replica < 0 || preAccept->Instance < 0){
        fprintf(stdout, "preAccept->Replica is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[preAccept->LeaderId][preAccept->Instance];

    if (preAccept->Seq >= maxSeq) {
        maxSeq = preAccept->Seq + 1;
    }

    if (inst != nullptr && (inst->status == COMMITTED || inst->status == ACCEPTED)) {
        if (inst->cmds.empty()) {
            inst->cmds = preAccept->Command;
            updateConflicts(preAccept->Command, preAccept->Replica, preAccept->Instance, preAccept->Seq);
        }
        // TODO
        // recordCommands(preAccept->cmds, preAccept->cmds_count);
        // sync(r);
        if(writeToDisk){
            fzxid = preAccept->LeaderId;
            fzxid = fzxid << 32;
            fzxid += preAccept->Instance;
            writeInstanceToFile(inst,fzxid);
        }
        //sync
        return;
    }
    if (preAccept->Instance >= crtInstance[preAccept->Replica]){
        crtInstance[preAccept->Replica] = preAccept->Instance + 1;
    }

    // update attributes for command
    int32_t seq = preAccept->Seq;
    std::array<int32_t, GROUP_SIZE> deps = preAccept->Deps;
    bool changed = updateAttributes(preAccept->Command, seq, deps, preAccept->Replica, preAccept->Instance);
    bool uncommittedDeps = false;

    for (int q = 0; q < group_size; q++) {
        if (deps[q] > committedUpTo[q]) {
            uncommittedDeps = true;
            break;
        }
    }
    STATUS status = PREACCEPTED_EQ;
    if (changed) {
        status = PREACCEPTED;
    }

    if (inst != nullptr) {
        if (preAccept->Ballot < inst->ballot) {
            PreAcceptReply * preacply= new PreAcceptReply(preAccept->Replica, preAccept->Instance, false,
                                                          inst->ballot, inst->seq, inst->deps, committedUpTo);
            if (preacply == nullptr) {
                fprintf(stderr, "Cannot alloc space for PreAcceptReply in handlePreAccept...\n");
                return;
            }
            replyPreAccept(preAccept->LeaderId, preacply);
//            if (Debug) {
//                fprintf(stdout, "I have replied to the PreAccept(%d.%d) by PreAcceptReply in pos1\n", preAccept->LeaderId, preAccept->Instance);
//            }
            delete preacply;
            preacply = nullptr;
            return;
        } else {
            inst->cmds = preAccept->Command;
            inst->seq = seq;
            inst->deps = deps;
            inst->ballot = preAccept->Ballot;
            inst->status = status;
        }
    } else {
        tk_instance * newInstance = new tk_instance(preAccept->Command, preAccept->Ballot,
                                                    status, seq, deps, nullptr);
        InstanceMatrix[preAccept->Replica][preAccept->Instance] = newInstance;
    }

    updateConflicts(preAccept->Command, preAccept->Replica, preAccept->Instance, preAccept->Seq);

    // TODO:: Record into stable storage
    // recordInstanceMetadata(InstanceSpace[preAccept->Replica][preAccept->Instance])
    // recordCommands(preAccept->Command)
    // sync()
    if(writeToDisk){
        fzxid = preAccept->LeaderId;
        fzxid = fzxid << 32;
        fzxid += preAccept->Instance;
        writeInstanceToFile(InstanceMatrix[preAccept->Replica][preAccept->Instance], fzxid);
    }
    //sync
    if (preAccept->Command.empty()) {
        latestCPReplica = preAccept->Replica;
        latestCPInstance = preAccept->Instance;
        // discard dependency hash tables
        clearHashTables();
    }

    if (changed || uncommittedDeps || preAccept->Replica != preAccept->LeaderId ||
        !isInitialBallot(preAccept->Ballot)) {
        PreAcceptReply * preacply= new PreAcceptReply(preAccept->Replica, preAccept->Instance, true,
                                                      preAccept->Ballot, seq, deps, committedUpTo);
        if (preacply == nullptr) {
            fprintf(stderr, "Cannot alloc space for PreAcceptReply in handlePreAccept...\n");
            return;
        }
        // TODO - DONE
        // replyPreAccept(preAccept->LeaderId, preacply);
        // TEST
        //cout<<"ready to reply pre accept "<<endl;
        replyPreAccept(preAccept->LeaderId, preacply);
//        DLOG_LOG(DEBUG,"3.ReplyPreAccept, Complete processing PreAccept msg, Command Leader: %d, Instance: %d",
//                 preAccept->LeaderId, preAccept->Instance);
        delete preacply;
        preacply = nullptr;
//        if (Debug) {
//            fprintf(stdout, "I have replied to the PreAccept(%d.%d) by PreAcceptReply in pos2\n", preAccept->LeaderId, preAccept->Instance);
//        }
    } else {
        PreAcceptOk * pok = new PreAcceptOk(preAccept->Instance);
        if (pok == nullptr) {
            fprintf(stderr, "Cannot alloc space for PreAcceptOK in handlePreAccept...\n");
            return;
        }
        // TODO - DONE
        // sendMsg(preAccept->LeaderId, pok);
        // TEST
        serializeBuff *buff = serialize_message(PREACCEPT_OK, (void *)pok);
        //cout<<"else ready to reply pre accept "<<endl;
        int ret = 0;
        if(UsingRDMA){
            ret = send_message(rdma_handlers[preAccept->LeaderId], buff, PREACCEPT_OK);
        }else{
            //pok->Marshal(Peers[preAccept->LeaderId]);
            send_message(Peers[preAccept->LeaderId],buff, PREACCEPT_OK);
        }

        if(ret == -1){
            Alive[preAccept->LeaderId] = false;
            destroyConnection(Peers[preAccept->LeaderId]);
            fprintf(stderr, "Fail Marshal PreAccept in handlePreAccept for replica %d\n", preAccept->LeaderId);
        }
        delete pok;
        pok = nullptr;
        delete buff;
        buff = nullptr;

    }
}


int Replica::reply_propose_ts_rdma(ProposeReplyTS * prts, rdma_fd *handler){
    int ret = 0;
    if(&handler == 0)
        return ret;
    serializeBuff *message = serialize_message(PROPOSE_REPLY_TS, (void *)prts);
    ret = send_message(handler, message, PROPOSE_REPLY_TS);
    delete message;
    return ret;
}

void Replica::handlePreAcceptReply(PreAcceptReply * pareply) {
    // TODO - DONE
    // LOG("Handling PreAccept Reply");
    // TEST
    if(pareply == nullptr){
        return;
    }
    if(pareply->Replica < 0 || pareply < 0){
        fprintf(stdout, "pareply->Replica is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[pareply->Replica][pareply->Instance];
    if (inst == nullptr || inst->status != PREACCEPTED) {
        // we have moved on, this is a delayed reply
        return;
    }

    if (inst->ballot != pareply->Ballot)
        return;

    if (!pareply->Ok) {
        // TODO: there is probably another active leader (original lost)
        inst->lb->nacks++;
        if (pareply->Ballot > inst->lb->maxRecvBallot)
            inst->lb->maxRecvBallot = pareply->Ballot;
        if (inst->lb->nacks >= group_size / 2) {
            // TODO (original lost)
        }
        return;
    }

    inst->lb->preAcceptOKs++;
    bool equal = mergeAttributes(inst->seq, inst->deps, pareply->Seq, pareply->Deps);
    if ((group_size <= 3 && !Thrifty) || inst->lb->preAcceptOKs > 1) {
        inst->lb->allEqual = inst->lb->allEqual & equal;
        if (!equal) {
            conflict++;
        }
    }
    bool allCommitted = true;
    for (int q = 0; q < group_size; q++) {
        if (inst->lb->committedDeps[q] < pareply->CommittedDeps[q])
            inst->lb->committedDeps[q] = pareply->CommittedDeps[q];
        if (inst->lb->committedDeps[q] < committedUpTo[q])
            inst->lb->committedDeps[q] = committedUpTo[q];
        if (inst->lb->committedDeps[q] < inst->deps[q])
            allCommitted = false;
    }

    // can we commit on the fast path?
    if (inst->lb->preAcceptOKs >= group_size * 3 / 4 &&
        inst->lb->allEqual && allCommitted &&
        isInitialBallot(inst->ballot)) {
        happy++;

 //       DLOG_LOG(DEBUG, "Fast path for instance %d.%d\n", pareply->Replica, pareply->Instance);
        // TODO - DONE
        // LOG("Fast path for instance %d.%d\n", pareply->Replica, pareply->Instance);
        // TEST
        // fprintf(stdout, "Fast Path For Instance %d.%d\n", pareply->Replica, pareply->Instance);
        InstanceMatrix[pareply->Replica][pareply->Instance]->status = COMMITTED;
        updateCommitted(pareply->Replica);
        if (!inst->lb->clientProposals.empty() && !Dreply) {
            // give clients the all clear
            for (int i = 0; i < inst->lb->clientProposals.size(); i++) {
                ProposeReplyTS * prts = new ProposeReplyTS(true, inst->lb->clientProposals[i].CommandId, 0,
                                                           "", inst->lb->clientProposals[i].Timestamp);


                // if (prts == nullptr) {
                //     fprintf(stderr, "Cannot alloc space for ProposeReplyTS in handlePreAcceptReply...\n");
                //     return;
                // }
                // TODO - DONE
                // replyProposeTS(prts, inst->lb->clientProposals[i].Reply);
                // here Reply is the handle of RDMA
                
                int ret = reply_propose_ts_rdma(prts, client_connections[inst->lb->clientProposals[i].Conn]);

                if(ret == -1){
                    printf("reply to client failed\n");
                }else{
                    printf("reply to client\n");
                }
                // if (!prts->Marshal(inst->lb->clientProposals[i].Conn)) {
                //     destroyConnection(inst->lb->clientProposals[i].Conn);
                //     fprintf(stderr, "Error when replying to client (socket %d)\n", inst->lb->clientProposals[i].Conn);
                // }
                delete prts;
                prts = nullptr;
//                if (Debug)
//                    fprintf(stdout, "Proposal %d Reply to client in handlePreAcceptReply\n", prts->CommandId);
            }
        }

        // TODO: record into stable storage
        // recordInstanceMetadata(inst);
        // sync();   // is this necessary here?
        if(writeToDisk){
            fzxid = pareply->Replica;
            fzxid = fzxid << 32;
            fzxid += pareply->Instance;
            writeStateToFile(inst,fzxid);
        }
        bcastCommit(pareply->Replica, pareply->Instance, inst->cmds, inst->seq, inst->deps);
 //       DLOG_LOG(DEBUG,"BroadCast commit message of Replica %d, Instance: %d", pareply->Replica, pareply->Instance);
    } else if (inst->lb->preAcceptOKs >= group_size / 2) {
        if (!allCommitted)
            weird++;
        slow++;
        inst->status = ACCEPTED;
        bcastAccept(pareply->Replica, pareply->Instance, inst->ballot,
                    int32_t(inst->cmds.size()), inst->seq, inst->deps);
    }
    // TODO: take the slow path if messages are slow to arrive (original lost)
//    if (Debug) {
//        fprintf(stdout, "Finish handling PreAcceptReply for instance %d\n", pareply->Instance);
//    }

//    delete pareply;
//    pareply = nullptr;
}

void Replica::handlePreAcceptOK(PreAcceptOk * preacok) {
    // TODO
    // LOG("Handling PreAccept OK\n");
    if(preacok == nullptr){
        fprintf(stdout, "PreAcceptOk is nullptr\n");
        return;
    }
    if(preacok->Instance < 0){
        fprintf(stdout, "preacok->Instance is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[Id][preacok->Instance];

    if (inst == nullptr || inst->status != PREACCEPTED) {
        // we have moved on, this is a delayed reply
        //DLOG_LOG(ERROR, "It's a delayed Reply, Intance: %d, Please Check~", preacok->Instance);
        return;
    }

    if (!isInitialBallot(inst->ballot))
        return;

    inst->lb->preAcceptOKs++;

    bool allCommitted = true;
    for (int q = 0; q < group_size; q++) {
        if (inst->lb->committedDeps[q] < inst->lb->originalDeps[q])
            inst->lb->committedDeps[q] = inst->lb->originalDeps[q];
        if (inst->lb->committedDeps[q] < committedUpTo[q])
            inst->lb->committedDeps[q] = committedUpTo[q];
        if (inst->lb->committedDeps[q] < inst->deps[q])
            allCommitted = false;
    }

//    if (Debug)
//        fprintf(stdout, "in handlePreAcceptOK: preAcceptOKs = %d, allCommitted = %d, initialBallot = %d\n",
//                inst->lb->preAcceptOKs, allCommitted, isInitialBallot(inst->ballot));

    // can we commit on the fast path?
    //printf("the preacceptok is = %d\n",inst->lb->preAcceptOKs);
    if (inst->lb->preAcceptOKs >= group_size / 2 &&
        inst->lb->allEqual && allCommitted &&
        isInitialBallot(inst->ballot)) {
        happy++;
        InstanceMatrix[Id][preacok->Instance]->status = COMMITTED;
        //printf("ready to update committed\n");
        updateCommitted(Id);
        if (!inst->lb->clientProposals.empty() && !Dreply) {
            // give clients the all clear
            for (int i = 0; i < inst->lb->clientProposals.size(); i++) {
                ProposeReplyTS * prts = new ProposeReplyTS(true, inst->lb->clientProposals[i].CommandId, 0,
                                                           "", inst->lb->clientProposals[i].Timestamp);
                int ret = reply_propose_ts_rdma(prts, client_connections[inst->lb->clientProposals[i].Conn]);

                if(ret == -1){
                    printf("reply to client failed\n");
                }else{
                    //printf("reply to client\n");
                }
                // TODO - DONE
                // replyProposeTS(prts, inst->lb->clientProposals[i].Reply);
                // here Reply is the handle of RDMA
                // if (!prts->Marshal(inst->lb->clientProposals[i].Conn)) {
                //     destroyConnection(inst->lb->clientProposals[i].Conn);
                //     fprintf(stderr, "Error when replying to client (socket %d)\n",inst->lb->clientProposals[i].Conn);
                // }

                delete prts;
                prts = nullptr;
//                if (Debug)
//                    fprintf(stdout, "Proposal %d Reply to client in handlePreAcceptOk\n", prts->CommandId);
            }
        }
        // TODO: record into stable storage
        // recordInstanceMetadata(inst);
        // sync();   // is this necessary here?
        if(writeToDisk){
            fzxid = Id;
            fzxid = fzxid << 32;
            fzxid += preacok->Instance;
            writeStateToFile(inst,fzxid);
        }
        
        bcastCommit(Id, preacok->Instance, inst->cmds, inst->seq, inst->deps);
//        DLOG_LOG(DEBUG,"BroadCast commit message of Replica %d, Instance: %d", Id, preacok->Instance);
    } else if (inst->lb->preAcceptOKs >= group_size / 2) {
        if (!allCommitted){
            weird++;
        }
        slow++;
        inst->status = ACCEPTED;
       //printf("hahaha\n");
        bcastAccept(Id, preacok->Instance, inst->ballot,
                    int32_t(inst->cmds.size()), inst->seq, inst->deps);
    }
    // TODO: take the slow path if messages are slow to arrive (original lost)
//    if (Debug) {
//        fprintf(stdout, "Finish handling PreAcceptOK for instance %d\n", preacok->Instance);
//    }

//    delete preacok;
//    preacok = nullptr;
}

/***************************************************************************************
 *                                   PHASE  2                                          *
 **************************************************************************************/

void Replica::handleAccept(Accept * accept) {
    if(accept == nullptr){
        fprintf(stdout, "accept is nullptr\n");
        return;
    }
    if(accept->Replica < 0 || accept->Instance < 0){
        fprintf(stdout, "accept->Replica is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[accept->Replica][accept->Instance];
    if (accept->Seq >= maxSeq)
        maxSeq = accept->Seq + 1;

    if (inst != nullptr && (inst->status == COMMITTED || inst->status == EXECUTED))
        return;

    if (accept->Instance >= crtInstance[accept->LeaderId])
        crtInstance[accept->LeaderId] = accept->Instance + 1;

    if (inst != nullptr) {
        if (accept->Ballot < inst->ballot) {
            AcceptReply * acr = new AcceptReply(accept->Replica, accept->Instance, inst->ballot, false);
            if (acr == nullptr) {
                fprintf(stderr, "Cannot alloc space for AcceptReply in handleAccept...\n");
                return;
            }
            replyAccept(accept->LeaderId, acr);
            delete acr;
            acr = nullptr;
            return;
        }
        inst->status = ACCEPTED;
        inst->seq = accept->Seq;
        inst->deps = accept->Deps;
    } else {
        tk_instance * newInstance = new tk_instance(emptyVec_cmd, accept->Ballot, ACCEPTED,
                                                    accept->Seq, accept->Deps, nullptr);
        InstanceMatrix[accept->Replica][accept->Instance] = newInstance;
        if (accept->Count == 0) {
            // checkpoint, update last checkpoint info
            latestCPReplica = accept->Replica;
            latestCPInstance = accept->Instance;
            clearHashTables();
        }
    }
    // TODO
    // recordInstanceMetadata(InstanceMatrix[accept->Replica][accept->Instance]);
    // sync();
    
    if(writeToDisk){
        fzxid = accept->Replica;
        fzxid = fzxid << 32;
        fzxid += accept->Instance;
        writeStateToFile(InstanceMatrix[accept->Replica][accept->Instance],fzxid);
    }
    AcceptReply * acr = new AcceptReply(accept->Replica, accept->Instance, accept->Ballot, true);
    if (acr == nullptr) {
        fprintf(stderr, "Cannot alloc space for AcceptReply in handleAccept...\n");
        return;
    }
    replyAccept(accept->LeaderId, acr);
    delete acr;
    acr = nullptr;
//    if (Debug) {
//        fprintf(stdout, "Finish handling Accept for replica %d's instance %d\n", accept->Replica, accept->Instance);
//    }

//    delete accept;
//    accept = nullptr;
}

void Replica::handleAcceptReply(AcceptReply * acreply) {
    if(acreply->Replica < 0 || acreply->Instance < 0){
        fprintf(stdout, "acreply->Replica is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[acreply->Replica][acreply->Instance];

    if (inst == nullptr || inst->status != ACCEPTED) {
        // we have moved on, they are delayed replies, so just ignore
        return;
    }
    if (!acreply->Ok) {
        // TODO: there is probably another active leader (original lost)
        inst->lb->nacks++;
        if (acreply->Ballot > inst->lb->maxRecvBallot)
            inst->lb->maxRecvBallot = acreply->Ballot;
        if (inst->lb->nacks >= group_size / 2) {
            // TODO (original lost)
        }
        return;
    }

    inst->lb->acceptOKs++;
    if (inst->lb->acceptOKs >= group_size / 2) {
        InstanceMatrix[acreply->Replica][acreply->Instance]->status = COMMITTED;
        updateCommitted(acreply->Replica);
        if (!inst->lb->clientProposals.empty() && !Dreply) {
            // give clients the all clear
            for (int i = 0; i < inst->lb->clientProposals.size(); i++) {
                ProposeReplyTS * prts = new ProposeReplyTS(true, inst->lb->clientProposals[i].CommandId, 0,
                                                           "", inst->lb->clientProposals[i].Timestamp);
                // if (prts == nullptr) {
                //     fprintf(stderr, "Cannot alloc space for ProposeReplyTS in handleAcceptReply...\n");
                //     return;
                // }
                int ret = reply_propose_ts_rdma(prts, client_connections[inst->lb->clientProposals[i].Conn]);

                if(ret == -1){
                    printf("reply to client failed\n");
                }
                // TODO - DONE
                // replyProposeTS(prts, inst->lb->clientProposals[i].Reply);
                // here Reply is the handle of RDMA
                // TEST
                // if (!prts->Marshal(inst->lb->clientProposals[i].Conn) ) {
                //     destroyConnection(inst->lb->clientProposals[i].Conn);
                //     fprintf(stderr, "Error when replying to client (socket %d)\n",inst->lb->clientProposals[i].Conn);
                // }
                delete prts;
                prts = nullptr;
//                if (Debug)
//                    fprintf(stdout, "Proposal %d Reply to client in handleAcceptReply\n", prts->CommandId);
            }
        }
        // TODO: record into stable storage
        // recordInstanceMetadata(inst);
        // sync();   // is this necessary here?
        if(writeToDisk){
            fzxid = acreply->Replica;
            fzxid = fzxid << 32;
            fzxid += acreply->Instance;
            writeStateToFile(inst,fzxid);
        }
        bcastCommit(acreply->Replica, acreply->Instance, inst->cmds, inst->seq, inst->deps);
    } else if (Debug) {
        fprintf(stdout, "have not received enough reply oks.\n");
    }
//    if (Debug) {
//        fprintf(stdout, "Finish handling AcceptReply for instance %d\n", acreply->Instance);
//    }

//    delete acreply;
//    acreply = nullptr;
}

/***************************************************************************************
 *                                 Commit Phase                                        *
 **************************************************************************************/

void Replica::handleCommit(Commit * commit) {
    if(commit == nullptr){
        fprintf(stdout, "commit msg is null\n");
        return;
    }
    if(commit->Replica < 0 || commit->Instance < 0){
        fprintf(stdout, "commit->Replica or commit->Instance is -1\n");
        return;
    }
    tk_instance * inst = InstanceMatrix[commit->Replica][commit->Instance];

    if (commit->Seq > maxSeq)
        maxSeq = commit->Seq + 1;
    if (commit->Instance >= crtInstance[commit->Replica])
        crtInstance[commit->Replica] = commit->Instance + 1;

    if (inst != nullptr) {
        if (inst->lb != nullptr && !inst->lb->clientProposals.empty() && commit->Command.empty()) {
            //someone committed a NO-OP, but we have proposals for this instance
            //try in a different instance
            for (Propose p: inst->lb->clientProposals) {
                Propose * pp = nullptr;
                try {
                    pp = new Propose(p);
                } catch (std::bad_alloc & bc) {
                    std::cout << "Cannot alloc space for Propose in handleCommit, " << bc.what() << std::endl;
                    return;
                }
                pro_mq->put(&pp);
                /* TODO: verify whether this implementation is right, here the original
                  implementation is r.ProposeChan <- p                               */
            }
            delete inst->lb;
            inst->lb = nullptr;
        }
        inst->seq = commit->Seq;
        inst->deps = commit->Deps;
        inst->status = COMMITTED;
        if(writeToDisk){
            fzxid = commit->Replica;
            fzxid = fzxid << 32;
            fzxid += commit->Instance;
            writeInstanceToFile(inst,fzxid);
        }
        
    } else {
        tk_instance * newInstance = new tk_instance(commit->Command, 0, COMMITTED,
                                                    commit->Seq, commit->Deps, nullptr);
        InstanceMatrix[commit->Replica][commit->Instance] = newInstance;
        updateConflicts(commit->Command, commit->Replica, commit->Instance, commit->Seq);

        if (commit->Command.empty()) {
            // checkpoint, update last checkpoint info
            latestCPReplica = commit->Replica;
            latestCPInstance = commit->Instance;
            clearHashTables();
        }
        if(writeToDisk){
            fzxid = commit->Replica;
            fzxid = fzxid << 32;
            fzxid += commit->Instance;
            writeInstanceToFile(newInstance, fzxid);
        }
        
    }
    updateCommitted(commit->Replica);
    /* TODO: record into stable storage
     * recordInstanceMetadata(InstanceMatrix[commit->Replica][commit->Instance]);
	 * recordCommands(commit->Command);                      */
//    if (Debug) {
//        fprintf(stdout, "Finish handling Commit for replica %d's instance %d\n", commit->Replica, commit->Instance);
//    }

//    delete commit;
//    commit = nullptr;
}

void Replica::handleCommitShort(CommitShort * commit) {
    if(commit->Replica < 0 || commit->Instance < 0){
        fprintf(stdout, "commit->Replica is -1 \n");
        return;
    }
    tk_instance * inst = InstanceMatrix[commit->Replica][commit->Instance];

    if (commit->Instance >= crtInstance[commit->Replica])
        crtInstance[commit->Replica] = commit->Instance + 1;

    if (inst != nullptr) {
        if (inst->lb != nullptr && !inst->lb->clientProposals.empty()) {
            // try command in a different instance
            for (Propose p: inst->lb->clientProposals) {
                Propose * pp = new Propose(p);
                if (pp == nullptr) {
                    fprintf(stderr, "Cannot alloc space for Propose in handleCommitShort...\n");
                    return;
                }
                pro_mq->put(&pp);
                /* TODO: verify whether this implementation is right, here the original
                  implementation is r.ProposeChan <- p                               */
            }
            delete inst->lb;
            inst->lb = nullptr;
        }
        inst->seq = commit->Seq;
        inst->deps = commit->Deps;
        inst->status = COMMITTED;
    } else {
        tk_instance * newInstance = new tk_instance(emptyVec_cmd, 0, COMMITTED,
                                                    commit->Seq, commit->Deps, nullptr);
        InstanceMatrix[commit->Replica][commit->Instance] = newInstance;

        if (commit->Count == 0) {
                // checkpoint, update last checkpoint info
            latestCPReplica = commit->Replica;
            latestCPInstance = commit->Instance;
            clearHashTables();
        }
    }
    updateCommitted(commit->Replica);
    if(writeToDisk){
        fzxid = commit->Replica;
        fzxid = fzxid << 32;
        fzxid += commit->Instance;
        writeStateToFile(InstanceMatrix[commit->Replica][commit->Instance],fzxid);
    }
    /* TODO: record into stable storage
     * recordInstanceMetadata(InstanceMatrix[commit->Replica][commit->Instance]); */
//    if (Debug) {
//        fprintf(stdout, "Finish handling CommitShort for replica %d's instance %d\n", commit->Replica, commit->Instance);
//    }

//    delete commit;
//    commit = nullptr;
}

/***************************************************************************************
 *                                 Helper Functions                                    *
 **************************************************************************************/

bool Replica::updateAttributes(std::vector<tk_command> & cmds, int32_t &seq,
                               std::array<int32_t, GROUP_SIZE> & deps,
                               int32_t replica, int32_t instance) {
    bool changed = 0;
    for (int q = 0; q < group_size; q++) {
        if (Id != replica && q == replica)
            continue;
        for (int i = 0; i < cmds.size(); i++) {
            if(cmds[i].opcode == CREATESESSION){
                //
                vector<int32_t> sessionCreated;
                sessionCreated.push_back(replica);
                //create instance
                sessionCreated.push_back(instance);                    
                //max instance
                sessionCreated.push_back(instance);
                sessionInstance[cmds[i].sessionId] = sessionCreated;
                continue;
            }else if(cmds[i].opcode == CLOSESESSION){
                auto s = sessionInstance.find(cmds[i].sessionId);
                deps[q] = s->second.at(2);
                continue;
            }
            auto p = (conflicts[q]).find(cmds[i].key);
            if (p != conflicts[q].end()) {
                if (p->second > deps[q]) {
                    deps[q] = p->second;
                    if (seq <= InstanceMatrix[q][p->second]->seq) {
                        seq = InstanceMatrix[q][p->second]->seq + 1;
                    }
                    changed = true;
                    break;
                }
            }else {
                auto s = sessionInstance.find(cmds[i].sessionId);
                if(s != sessionInstance.end()){
                    if(q == s->second.at(0))
                        deps[q] = s->second.at(1);
                }
            }
        }
    }
    for (int i = 0; i < cmds.size(); i++) {
        auto s = maxSeqPerKey.find(cmds[i].key);
        if (s != maxSeqPerKey.end() && seq <= s->second) {
            changed = true;
            seq = s->second + 1;
        }
    }
    return changed;
}

bool Replica::mergeAttributes(int32_t & seq1, std::array<int32_t, GROUP_SIZE> & deps1,
                              int32_t seq2, std::array<int32_t, GROUP_SIZE> & deps2) {
    bool equal = true;
    if (seq1 != seq2) {
        equal = false;
        if (seq2 > seq1)
            seq1 = seq2;
    }
    for (int q = 0; q < group_size; q++) {
        if (q == Id) {
            continue;
        }
        if (deps1[q] != deps2[q]) {
            equal = false;
            if (deps2[q] > deps1[q])
                deps1[q] = deps2[q];
        }
    }
    return equal;
}

bool Replica::Conflict(tk_command &delta, tk_command &gamma) {
    if(delta.sessionId == gamma.sessionId){
        if(delta.opcode == CREATESESSION || gamma.opcode == CREATESESSION)
            return true;
    }
    if (delta.key == gamma.key) {
        if (delta.opcode == PUT || gamma.opcode == PUT)
            return true;
    }
    return false;
}

bool Replica::ConflictBatch(std::vector<tk_command> &batch1,
                            std::vector<tk_command> &batch2) {
    for (int i = 0; i < batch1.size(); i++) {
        for (int j = 0; j < batch2.size(); j++) {
            if (Conflict(batch1[i], batch2[j]))
                return true;
        }
    }
    return false;
}

void Replica::updateConflicts(std::vector<tk_command> &cmds, int32_t replica,
                              int32_t instance, int32_t seq) {
    for (int i = 0; i < cmds.size(); i++) {
        //if op = closesession
        if(cmds[i].opcode == CLOSESESSION){
            sessionInstance.erase(cmds[i].sessionId);
            continue;
        }
        //
        auto d = (conflicts[replica]).find(std::string(cmds[i].key));
        if (d != conflicts[replica].end() && d->second < instance) {
            conflicts[replica][cmds[i].key] = instance;
        } else {
            conflicts[replica][cmds[i].key] = instance;
        }
        auto s = maxSeqPerKey.find(cmds[i].key);
        if (s != maxSeqPerKey.end() && s->second < seq) {
            maxSeqPerKey[cmds[i].key] = seq;
        } else {
            maxSeqPerKey[cmds[i].key] = seq;
        }
    }
}

void Replica::updateCommitted(int32_t replica) {
    while (InstanceMatrix[replica][committedUpTo[replica]+1] != nullptr &&
           (InstanceMatrix[replica][committedUpTo[replica]+1]->status == COMMITTED ||
            InstanceMatrix[replica][committedUpTo[replica]+1]->status == EXECUTED)) {
        committedUpTo[replica]++;
    }
}

void Replica::updatePreferredPeerOrder(std::vector<int32_t> & quorum) {
    std::vector<int32_t> aux((unsigned long)group_size, 0);
    int i = 0;
    for (int32_t p: quorum) {
        if (p == Id)
            continue;
        aux[i++] = p;
    }
    for (int32_t p: PreferredPeerOrder) {
        bool found = false;
        for (int j = 0; j < i; j++) {
            if (aux[j] == p) {
                found = true;
                break;
            }
        }
        if (!found) {
            aux[i++] = p;
        }
    }
    PreferredPeerOrder = aux;
}

void Replica::clearHashTables() {
    for (int q = 0; q < group_size; q++) {
        conflicts[q].clear();
    }
}

int32_t Replica::makeUniqueBallot(int32_t ballot) {
    return (ballot << 4) | Id;
}

int32_t Replica::makeBallotLargerThan(int32_t ballot) {
    return makeUniqueBallot((ballot >> 4) + 1);
}


/*********************************************************************************************
 *                               Inter-Replica communication                                 *
 ********************************************************************************************/



void Replica::bcastPreAccept(int32_t replica, int32_t instance, uint32_t ballot,
                             std::vector<tk_command> &cmds, int32_t seq,
                             std::array<int32_t, GROUP_SIZE> & deps) {
//    if (Debug)
       // printf("Replica %d going to broadcast PreAccept for instance %d\n", replica, instance);
  //      gettimeofday(&tv_end, NULL);
   // thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
  //  cout<<"the latency before create preaccept "<<thistime<<endl;
    PreAccept *preAccept = nullptr;
    if(preaccept_queue.empty()){
        for(int i = 0; i < 1024; i++){
            preAccept = new PreAccept();
            preaccept_queue.push(preAccept);
        }

    }

    preAccept =preaccept_queue.front();
    preaccept_queue.pop();
    //gettimeofday(&tv_end, NULL);
    //thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
    //cout<<"the latency after create preaccept "<<thistime<<endl;
    preAccept->LeaderId = Id;
    preAccept->Replica = replica;
    preAccept->Ballot = ballot;
    preAccept->Instance = instance;
    preAccept->Command = cmds;
    preAccept->Seq = seq;
    preAccept->Deps = deps;
//    pa.Command.clear();
//    pa.LeaderId = Id; pa.Replica = replica; pa.Ballot = ballot;
//    pa.Instance = instance; pa.Command = cmds; pa.Seq = seq; pa.Deps = deps;
    int n = group_size - 1, sent = 0;
    if (Thrifty) {
        n = group_size * 3 / 4 + 1; // original group_size / 2;
    }

    
    // printf("[TIME From recv to bc]: %lf us\n", (double) diff.count() * 1000000);
   // printf("start going to serialize preaccept command size = %d\n",cmds.size());
   // gettimeofday(&tv_end, NULL);
    //thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
   // cout<<"the latency before a serialize "<<thistime<<endl;
    serializeBuff *msg = serialize_message(PREACCEPT, (void *)preAccept);
   // gettimeofday(&tv_end, NULL);
   // thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
    //cout<<"the latency after a serialize "<<thistime<<endl;
   
//    std::vector< std::future<int> > results;
//    std::shared_ptr<serializeBuff> msg_s(msg);
    //cout<<"msg off"<<msg->off<<endl;
    
    //printf("bcast preaccept groupsize %d\n",group_size);
    //for (int q = 0; q < 2; q++) {
    for (int q = 0; q < group_size; q++) {
        // if (!Alive[PreferredPeerOrder[q]]) {
        //     fprintf(stderr, "Pass replica %d since disconnected when bcastPreAccept\n", PreferredPeerOrder[q]);
        //     continue;
        // }
        if(q == Id){
            continue;
        }
        // TODO - DONE
        // sendPreAccept(PreferredPeerOrder[q], &pa);
        // if this method fails, log failure
        // TEST
        //fprintf(stdout, "Send to replica %d\n", q);
        //count()
        if(UsingRDMA){
            int ret = send_message(rdma_handlers[q], msg, PREACCEPT);   
        }else{
            if (!Alive[PreferredPeerOrder[q]])
            continue;
            //preAccept->Marshal(Peers[PreferredPeerOrder[q]]);
            send_message(Peers[PreferredPeerOrder[q]], msg, PREACCEPT); 
        }

        sent++;
//        DLOG_LOG(DEBUG, "Replica %d have sent PreAccept of Instance Id: %d , pa Instance Id: %d to %d completely",
//                 replica, instance, preAccept->Instance, PreferredPeerOrder[q]);
        if (sent >= n){
            break;
        }

    }
    // if(ret == -1){
    //         //destroyConnection(Peers[PreferredPeerOrder[q]]);
    //         fprintf(stdout, "replica %d is not alive when bcastPreAccept\n", q);
    //         //Alive[PreferredPeerOrder[q]] = false;
    //         Alive[q] = false;
    // }
    // for(auto && ret :  results){
    //     if()
    // }

//    delete msg;
    send_buff_queue.push(msg);
    delete preAccept;

//    if (Debug)
//        printf("Replica %d finish broadcast PreAccept for instance %d\n", replica, instance);
}

void Replica::bcastCommit(int32_t replica, int32_t instance, std::vector<tk_command> &cmds,
                          int32_t seq, std::array<int32_t, GROUP_SIZE> &deps) {
//    if (Debug)
//        printf("Replica %d going to broadcast commit for instance %d\n", replica, instance);
    //cm.Command.clear();

    cm.LeaderId = Id; cm.Replica = replica; cm.Instance = instance;
    cm.Seq = seq; cm.Command = cmds; cm.Deps = deps;
    cms.LeaderId = Id; cms.Replica = replica; cms.Instance = instance;
    cms.Count = (int32_t)cmds.size(); cms.Seq = seq; cms.Deps = deps;
    int sent = 0;
    serializeBuff *msg = serialize_message(COMMIT, (void *) &cm);
    serializeBuff *msg1 = serialize_message(COMMIT_SHORT, (void *) &cms);
    for (int q = 0; q < group_size; q++) {
        if(q == Id){
            continue;
        }
        // if (!Alive[PreferredPeerOrder[q]]) {
        //     fprintf(stderr, "Pass replica %d since disconnected when bcastCommit\n", PreferredPeerOrder[q]);
        //     continue;
        // }
        if (Thrifty && sent > group_size * 3 / 4) {
            // TODO - DONE
            // sendCommit(PreferredPeerOrder[q], &cm);
            // if this or the next method fails, log failure
            // TEST
            if(UsingRDMA){
                int ret = send_message(rdma_handlers[q], msg, COMMIT);
            }else{
                if (!Alive[PreferredPeerOrder[q]])
                    continue;
                //cm.Marshal(Peers[PreferredPeerOrder[q]]);
                send_message(Peers[PreferredPeerOrder[q]], msg, COMMIT);
            }
            
//            send_message_pool->enqueue(send_message, rdma_handlers[q], msg, COMMIT);
            // if(ret == -1){
            //     //destroyConnection(Peers[PreferredPeerOrder[q]]);
            //     fprintf(stdout, "replica %d is not alive when bcastCommit\n", q);
            //     Alive[q] = false;
            //     //Alive[PreferredPeerOrder[q]] = false;
            // }

        } else {
            // TODO - DONE
            // sendCommitShort(PreferredPeerOrder[q], &cms);
           if(UsingRDMA){
                int ret = send_message(rdma_handlers[q], msg1, COMMIT_SHORT);
           }else{
                if (!Alive[PreferredPeerOrder[q]])
                    continue;
                //cms.Marshal(Peers[PreferredPeerOrder[q]]);
                send_message(Peers[PreferredPeerOrder[q]], msg1, COMMIT_SHORT);
           }
            
//            send_message_pool->enqueue(send_message, rdma_handlers[q], msg1, COMMIT_SHORT);
           //  if(ret == -1){
           // //     destroyConnection(Peers[PreferredPeerOrder[q]]);
           //      fprintf(stdout, "replica %d is not alive when bcastCommit\n", q);
           // //     Alive[PreferredPeerOrder[q]] = false;
           //  }
//            send_message_pool->enqueue(send_message, rdma_handlers[q], msg, PREACCEPT);
//            
//            DLOG_LOG(DEBUG, "Replica %d have sent commit short of Instance Id: %d to %d completely", replica, instance, PreferredPeerOrder[q]);
            sent++;
        }
    }
    send_buff_queue.push(msg);
    send_buff_queue.push(msg1);
//    delete msg;
//    delete msg1;

//    if (Debug)
//        printf("Replica %d finish broadcast commit for instance %d\n", replica, instance);
}

void Replica::bcastAccept(int32_t replica, int32_t instance, int32_t ballot, int32_t count,
                          int32_t seq, std::array<int32_t, GROUP_SIZE> &deps) {
//    if (Debug)
//        printf("Replica %d going to broadcast accept for instance %d\n", replica, instance);
    cm.Command.clear();
    ea.LeaderId = Id; ea.Replica = replica; ea.Instance = instance; ea.Ballot = ballot;
    ea.Count = count; ea.Seq = seq; ea.Deps = deps;
    int n = group_size - 1;
    if (Thrifty)
        n = group_size / 2;

    int sent = 0;
    serializeBuff *msg = serialize_message(ACCEPT, (void *) &ea);

    for (int q = 0; q < group_size; q++) {
        // if (!Alive[PreferredPeerOrder[q]]) {
        //     fprintf(stderr, "Pass replica %d since disconnected when bcastAccept\n", PreferredPeerOrder[q]);
        //     continue;
        // }
        if(q == Id){
            continue;
        }
        // TODO - DONE
        // sendAccept(PreferredPeerOrder[q], &ea);
        // TEST
        int ret = 0;
//        send_message_pool->enqueue(send_message, rdma_handlers[q], msg, PREACCEPT);
        if(UsingRDMA){
            ret = send_message(rdma_handlers[q], msg, ACCEPT);
        }else{
            if (!Alive[PreferredPeerOrder[q]])
                continue;
            //ea.Marshal(Peers[PreferredPeerOrder[q]]);
            send_message(Peers[PreferredPeerOrder[q]], msg, ACCEPT);
        }

        if(ret == -1){
            //destroyConnection(Peers[PreferredPeerOrder[q]]);
            fprintf(stdout, "replica %d is not alive when bcastAccept\n", q);
            Alive[q] = false;
        }


//        
        sent++;
//        DLOG_LOG(DEBUG, "Replica %d have sent Accept of Instance Id: %d to %d completely", replica, instance, q);
        // if (sent >= n)
        //     break;
    }
    delete msg;

//    if (Debug)
//        printf("Replica %d finish broadcast accept for instance %d\n", replica, instance);
}

void Replica::bcastPrepare(int32_t replica, int32_t instance, int32_t ballot) {
    pr.LeaderId = Id; pr.Replica = replica; pr.Instance = instance; pr.Ballot = ballot;
    int n = group_size - 1;
    if (Thrifty)
        n = group_size / 2;
    int q = Id, sent = 0;
    serializeBuff* msg = serialize_message(PREPARE, (void *) &pr);
    while (sent < n) {
        q = (q + 1) % int32_t(group_size);
        if (q == Id) {
            // TODO - DONE
            // LOG("Not enough replicas alive");
            // TEST
            fprintf(stderr, "Not Enough Replicas Alive!\n");
            break;
        }
        if (!Alive[q])
            continue;
        // TODO - DONE
        // sendPrepare(q, &pr);
        // TEST
        int ret = 0;
        if(UsingRDMA){
            ret = send_message(rdma_handlers[q], msg, PREPARE);
        }else{
            if (!Alive[PreferredPeerOrder[q]])
                continue;
            //pr.Marshal(Peers[PreferredPeerOrder[q]]);
            send_message(Peers[PreferredPeerOrder[q]], msg, PREPARE);
        }
        if(ret == -1){
            Alive[q] = false;

        }
//        
//        DLOG_LOG(DEBUG, "Replica %d have sent PreAccept of Instance Id: %d to %d completely", replica, instance, q);
        sent++;
    }
    delete msg;
}

void Replica::bcastTryPreAccept(int32_t replica, int32_t instance, int32_t ballot,
                                std::vector<tk_command> &cmds,
                                int32_t seq, std::array<int32_t, GROUP_SIZE> &deps) {
    tpa.Command.clear();
    tpa.LeaderId = Id; tpa.Replica = replica; tpa.Instance = instance;
    tpa.Ballot = ballot; tpa.Seq = seq; tpa.Command = cmds; tpa.Deps = deps;
    serializeBuff *msg = serialize_message(TRY_PREACCEPT, (void *) &tpa);
    for (int q = 0; q < group_size; q++) {
        if (q == Id)
            continue;
        if (!Alive[q])
            continue;
        // TODO - DONE
        // sendTryPreAccept(q, &tpa);
        // TEST
        if(UsingRDMA){
            int ret = send_message(rdma_handlers[q], msg, TRY_PREACCEPT);
            if(ret == -1){
                Alive[q] = false;
            }
        }else{
            //tpa.Marshal(Peers[q]);
            send_message(Peers[PreferredPeerOrder[q]], msg, TRY_PREACCEPT);
        }
        

//        if (!tpa.Marshal(Peers[q]))
//            Alive[q] = false;
    }
    delete msg;
}

/*********************************************************************************************
 *                                     Recovery Actions                                      *
 ********************************************************************************************/

void Replica::startRecoveryForInstance(int32_t replica, int32_t instance) {
    std::array<int32_t, GROUP_SIZE> nildeps;

    if (InstanceMatrix[replica][instance] == nullptr) {
        tk_instance * newInstance = new tk_instance(emptyVec_cmd, 0, NONE,
                                                    0, nildeps, nullptr);
        InstanceMatrix[replica][instance] = newInstance;
    }

    tk_instance * inst = InstanceMatrix[replica][instance];
    if (inst->lb == nullptr) {
        lb_t * nlb = new lb_t(emptyVec_pro, -1, 0, false, 0, 0, 0, nildeps,
                              emptyVec_int, nullptr, true, false, emptyVec_bool, 0);
        inst->lb = nlb;
    } else {
        lb_t * nlb = new lb_t(inst->lb->clientProposals, -1, 0, false, 0, 0, 0, nildeps,
                              emptyVec_int, nullptr, true, false, emptyVec_bool, 0);
        inst->lb = nlb;
    }
    if (inst->status == ACCEPTED) {
        RecoverInstance * rvi = new RecoverInstance(false, inst->status, inst->seq,
                                                    0, inst->deps, inst->cmds);
        if (rvi == nullptr) {
            fprintf(stderr, "Cannot alloc space for RecoverInstance in startRecoveryForInstance...\n");
        }
        inst->lb->recoveryInst = rvi;
        inst->lb->maxRecvBallot = inst->ballot;
    } else if (inst->status == PREACCEPTED) {
        RecoverInstance * rvi = new RecoverInstance(Id == replica, inst->status, inst->seq,
                                                    1, inst->deps, inst->cmds);
        if (rvi == nullptr) {
            fprintf(stderr, "Cannot alloc space for RecoverInstance in startRecoveryForInstance...\n");
        }
        inst->lb->recoveryInst = rvi;
    }

    // compute large ballot
    inst->ballot = makeBallotLargerThan(inst->ballot);
    bcastPrepare(replica, instance, inst->ballot);
}

void Replica::handlePrepare(Prepare *prepare) {
    tk_instance * inst = InstanceMatrix[prepare->Replica][prepare->Instance];
    PrepareReply * preply;
    std::array<int32_t, GROUP_SIZE> nildeps;

    if (inst == nullptr) {
        tk_instance * newInstance = new tk_instance(emptyVec_cmd, prepare->Ballot,
                                                    NONE, 0, nildeps, nullptr);
        preply = new PrepareReply(Id, prepare->Replica, prepare->Instance, true, -1,
                                  NONE, emptyVec_cmd, -1, nildeps);
    } else {
        bool ok = true;
        if (prepare->Ballot < inst->ballot)
            ok = false;
        else
            inst->ballot = prepare->Ballot;
        preply = new PrepareReply(Id, prepare->Replica, prepare->Instance, ok, inst->ballot,
                                  inst->status, inst->cmds, inst->seq, inst->deps);
    }
    if (preply == nullptr) {
        fprintf(stderr, "Cannot alloc space for PrepareReply in handlePrepare...\n");
        return;
    }
    replyPrepare(prepare->LeaderId, preply);
}

void Replica::handlePrepareReply(PrepareReply * preply) {
    tk_instance * inst = InstanceMatrix[preply->Replica][preply->Instance];

    if (inst == nullptr || inst->lb == nullptr || !inst->lb->preparing) {
        // we've moved on -- these are delayed replies, so just ignore
        // TODO: should replies for non-current ballots be ignored? (original lost)
        return;
    }

    if (!preply->Ok) {
        // TODO: there is probably another active leader, back off and retry later (original lost)
        inst->lb->nacks++;
        return;
    }

    //Got an ACK (preply.OK == TRUE)

    inst->lb->prepareOKs++;

    if (preply->Status == COMMITTED || preply->Status == EXECUTED) {
        tk_instance * newInstance = new tk_instance(preply->Command, inst->ballot, COMMITTED,
                                                    preply->Seq, preply->Deps, nullptr);
        InstanceMatrix[preply->Replica][preply->Instance] = newInstance;
        bcastCommit(preply->Replica, preply->Instance, inst->cmds, preply->Seq, preply->Deps);
        //TODO: check if we should send notifications to clients (original lost)
        return;
    }

    if (preply->Status == ACCEPTED) {
        if (inst->lb->recoveryInst == nullptr || inst->lb->maxRecvBallot < preply->Ballot) {
            RecoverInstance * rvi = new RecoverInstance(false, preply->Status, 0, preply->Seq,
                                                        preply->Deps, preply->Command);
            if (rvi == nullptr) {
                fprintf(stderr, "Cannot alloc space for RevoerInstance in handlePrePareReply...\n");
            }
            inst->lb->recoveryInst = rvi;
            inst->lb->maxRecvBallot = preply->Ballot;
        }
    }

    if ((preply->Status == PREACCEPTED || preply->Status == PREACCEPTED_EQ) &&
        (inst->lb->recoveryInst == nullptr || inst->lb->recoveryInst->status < ACCEPTED)) {
        if (inst->lb->recoveryInst == nullptr) {
            RecoverInstance * rvi = new RecoverInstance(false, preply->Status, 1, preply->Seq,
                                                        preply->Deps, preply->Command);
            if (rvi == nullptr) {
                fprintf(stderr, "Cannot alloc space for RevoerInstance in handlePrePareReply...\n");
            }
            inst->lb->recoveryInst = rvi;
        } else if (preply->Seq == inst->seq && equal(preply->Deps, inst->deps)) {
            inst->lb->recoveryInst->preAcceptCount++;
        } else if (preply->Status == PREACCEPTED_EQ) {
            // If we get different ordering attributes from pre-acceptors, we must go with the ones
            // that agreed with the initial command leader (in case we do not use Thrifty).
            // This is safe if we use thrifty, although we can also safely start phase 1 in that case.
            RecoverInstance * rvi = new RecoverInstance(false, preply->Status, 1, preply->Seq,
                                                        preply->Deps, preply->Command);
            if (rvi == nullptr) {
                fprintf(stderr, "Cannot alloc space for RevoerInstance in handlePrePareReply...\n");
            }
            inst->lb->recoveryInst = rvi;
        }
        if (preply->AcceptorId == preply->Replica) {
            //if the reply is from the initial command leader, then it's safe to restart phase 1
            inst->lb->recoveryInst->leaderResponded = true;
            return;
        }
    }

    if (inst->lb->prepareOKs < group_size / 2)
        return;

    //Received Prepare replies from a majority

    RecoverInstance * ir = inst->lb->recoveryInst;

    if (ir != nullptr) {
        //at least one replica has (pre-)accepted this instance
        if (ir->status == ACCEPTED ||
            (!ir->leaderResponded && ir->preAcceptCount >= group_size / 2 &&
             (Thrifty || ir->status == PREACCEPTED_EQ))) {
            //safe to go to Accept phase
            inst->cmds = ir->cmds;
            inst->seq = ir->seq;
            inst->deps = ir->deps;
            inst->status = ACCEPTED;
            inst->lb->preparing = false;
            bcastAccept(preply->Replica, preply->Instance, inst->ballot,
                        int32_t(inst->cmds.size()), inst->seq, inst->deps);
        } else if (!ir->leaderResponded && ir->preAcceptCount >= (group_size/2 + 1)/2) {
            //send TryPreAccepts
            //but first try to pre-accept on the local replica
            inst->lb->preAcceptOKs = 0;
            inst->lb->nacks = 0;
            inst->lb->possibleQuorum = std::vector<bool>((unsigned long)group_size, false);
            for (int q = 0; q < group_size; q++) {
                inst->lb->possibleQuorum[q] = true;
            }
            int q, i;
            if (findPreAcceptConflicts(ir->cmds, preply->Replica, preply->Instance, ir->seq, ir->deps, &q, &i)) {
                if (q == -1 || i == -1)
                    return;
                if (InstanceMatrix[q][i]->status >= COMMITTED) {
                    //start Phase1 in the initial leader's instance
                    startPhase1(preply->Replica, preply->Instance, inst->ballot,
                                inst->lb->clientProposals, ir->cmds, ir->cmds.size());
                    return;
                } else {
                    inst->lb->nacks = 1;
                    inst->lb->possibleQuorum[Id] = false;
                }
            } else {
                inst->cmds = ir->cmds;
                inst->seq = ir->seq;
                inst->deps = ir->deps;
                inst->status = PREACCEPTED;
                inst->lb->preAcceptOKs = 1;
            }
            inst->lb->preparing = false;
            inst->lb->tryingToPreAccept = true;
            bcastTryPreAccept(preply->Replica, preply->Instance, inst->ballot,
                              inst->cmds, inst->seq, inst->deps);
        } else {
            //start Phase1 in the initial leader's instance
            inst->lb->preparing = false;
            startPhase1(preply->Replica, preply->Instance, inst->ballot,
                        inst->lb->clientProposals, ir->cmds, ir->cmds.size());
        }
    } else {
        //try to finalize instance by proposing NO-OP
        std::array<int32_t, GROUP_SIZE> noop_deps;
        // commands that depended on this instance must look at all previous instances
        noop_deps[preply->Replica] = preply->Instance - 1;
        inst->lb->preparing = false;
        tk_instance * newInstance = new tk_instance(emptyVec_cmd, inst->ballot, ACCEPTED, 0, noop_deps, inst->lb);
        InstanceMatrix[preply->Replica][preply->Instance] = newInstance;
        bcastAccept(preply->Replica, preply->Instance, inst->ballot, 0, 0, noop_deps);
    }
}

void Replica::handleTryPreAccept(TryPreAccept * tpap) {
    tk_instance * inst = InstanceMatrix[tpap->Replica][tpap->Instance];

    if (inst != nullptr && inst->ballot > tpap->Ballot) {
        // ballot number too small
        TryPreAcceptReply * ntpap = new TryPreAcceptReply(Id, tpap->Replica, tpap->Instance, inst->ballot,
                                                          false, tpap->Replica, tpap->Instance, inst->status);
        if (ntpap != nullptr) {
            fprintf(stderr, "Cannot alloc space for TryPreAcceptReply in handleTryPreAccept...\n");
            return;
        }
        replyTryPreAccept(tpap->LeaderId, ntpap);
    }
    int confRep, confInst;
    if (findPreAcceptConflicts(tpap->Command, tpap->Replica, tpap->Instance,
                               tpap->Seq, tpap->Deps, &confRep, &confInst)) {
        // there is a conflict, can't pre-accept
        TryPreAcceptReply * ntpap = new TryPreAcceptReply(Id, tpap->Replica, tpap->Instance, inst->ballot,
                                                          false, confRep, confInst,
                                                          InstanceMatrix[confRep][confInst]->status);
        if (ntpap != nullptr) {
            fprintf(stderr, "Cannot alloc space for TryPreAcceptReply in handleTryPreAccept...\n");
            return;
        }
        replyTryPreAccept(tpap->LeaderId, ntpap);
    } else {
        // can pre-accept
        if (tpap->Instance >= crtInstance[tpap->Replica]) {
            crtInstance[tpap->Replica] = tpap->Instance + 1;
        }
        if (inst != nullptr) {
            inst->cmds = tpap->Command;
            inst->deps = tpap->Deps;
            inst->seq = tpap->Seq;
            inst->status = PREACCEPTED;
            inst->ballot = tpap->Ballot;
        } else {
            tk_instance * newInstance = new tk_instance(tpap->Command, tpap->Ballot, PREACCEPTED,
                                                        tpap->Seq, tpap->Deps, nullptr);
            InstanceMatrix[tpap->Replica][tpap->Instance] = newInstance;
        }
        TryPreAcceptReply * ntpap = new TryPreAcceptReply(Id, tpap->Replica, tpap->Instance,
                                                          inst->ballot, true, 0, 0, NONE);
        if (ntpap != nullptr) {
            fprintf(stderr, "Cannot alloc space for TryPreAcceptReply in handleTryPreAccept...\n");
            return;
        }
        replyTryPreAccept(tpap->LeaderId, ntpap);
    }
}

void Replica::handleTryPreAcceptReply(TryPreAcceptReply * tpar) {
    tk_instance * inst = InstanceMatrix[tpar->Replica][tpar->Instance];
    if (inst == nullptr || inst->lb == nullptr ||
        !inst->lb->tryingToPreAccept ||
        inst->lb->recoveryInst == nullptr) {
        return;
    }

    RecoverInstance * ir = inst->lb->recoveryInst;

    if (tpar->Ok) {
        inst->lb->preAcceptOKs++;
        inst->lb->tpaOks++;
        if (inst->lb->preAcceptOKs >= group_size / 2) {
            //it's safe to start Accept phase
            inst->cmds = ir->cmds;
            inst->seq = ir->seq;
            inst->deps = ir->deps;
            inst->status = ACCEPTED;
            inst->lb->tryingToPreAccept = false;
            inst->lb->acceptOKs = 0;
            bcastAccept(tpar->Replica, tpar->Instance, inst->ballot, int32_t(inst->cmds.size()), inst->seq, inst->deps);
            return;
        }
    } else {
        inst->lb->nacks++;
        if (tpar->Ballot > inst->ballot) {
            // TODO: retry with higher ballot (original lost)
            return;
        }
        inst->lb->tpaOks++;
        if (tpar->ConflictReplica == tpar->Replica && tpar->ConflictInstance == tpar->Instance) {
            // TODO: re-run prepare (original lost)
            inst->lb->tryingToPreAccept = false;
            return;
        }
        inst->lb->possibleQuorum[tpar->AcceptorId] = false;
        inst->lb->possibleQuorum[tpar->ConflictReplica] = false;

        int notInQuorum = 0;
        for (int q = 0; q < group_size; q++) {
            if (!inst->lb->possibleQuorum[tpar->AcceptorId]) {
                notInQuorum++;
            }
        }
        if (tpar->ConflictStatus >= COMMITTED || notInQuorum > group_size / 2) {
            // abandon recovery, restart from phase 1
            inst->lb->tryingToPreAccept = false;
            startPhase1(tpar->Replica, tpar->Instance, inst->ballot,
                        inst->lb->clientProposals, ir->cmds, ir->cmds.size());
        }
        if (notInQuorum == group_size / 2) {
            // this is to prevent defer cycles
            int32_t dq, di;
            if (deferredByInstance(tpar->Replica, tpar->Instance, &dq, &di)) {
                if (inst->lb->possibleQuorum[dq]) {
                    // an instance whose leader must have been in this instance's quorum has been deferred for
                    // this instance => contradiction abandon recovery, restart from phase 1
                    inst->lb->tryingToPreAccept = false;
                    startPhase1(tpar->Replica, tpar->Instance, inst->ballot,
                                inst->lb->clientProposals, ir->cmds, ir->cmds.size());
                }
            }
        }
        if (inst->lb->tpaOks >= group_size / 2) {
            //defer recovery and update deferred information
            updateDeferred(tpar->Replica, tpar->Instance, tpar->ConflictReplica, tpar->ConflictInstance);
            inst->lb->tryingToPreAccept = false;
        }
    }
}

bool Replica::findPreAcceptConflicts(std::vector<tk_command> &cmds, int32_t replica,
                                     int32_t instance, int32_t seq,
                                     std::array<int32_t, GROUP_SIZE> &deps,
                                     int *qr, int *ir) {
    if(replica < 0 || instance < 0){
        fprintf(stdout, "error instance or replica is -1\n");
        return false;
    }
    tk_instance * inst = InstanceMatrix[replica][instance];
    if (inst != nullptr && inst->cmds.size() > 0) {
        if (inst->status >= ACCEPTED) {
            // already ACCEPTED or COMMITTED
            // we consider this a conflict because we shouldn't regress to PRE-ACCEPTED
            *qr = replica; *ir = instance;
            return true;
        }
        if (inst->seq == tpa.Seq && equal(inst->deps, tpa.Deps)) {
            // already PRE-ACCEPTED, no point looking for conflicts again
            *qr = replica; *ir = instance;
            return false;
        }
    }
    for (int q = 0; q < group_size; q++) {
        for (int32_t i = executeUpTo[q]; i < crtInstance[q]; i++) {
            if (replica == q && instance == i) {
                // no point checking past instance in replica's row, since replica would have
                // set the dependencies correctly for anything started after instance
                break;
            }
            if (i == deps[q]) {
                //the instance cannot be a dependency for itself
                continue;
            }
            tk_instance * _inst = InstanceMatrix[q][i];
            if (_inst == nullptr || _inst->cmds.empty()) {
                continue;
            }
            if (_inst->deps[replica] >= instance) {
                // instance q.i depends on instance replica.instance, it is not a conflict
                continue;
            }
            if (ConflictBatch(_inst->cmds, cmds)) {
                if (i > deps[q] ||
                    (i < deps[q] && _inst->seq >= seq && (q != replica || _inst->status > PREACCEPTED_EQ))) {
                    // this is a conflict


                    *qr = q; *ir = i;
                    return true;
                }
            }
        }
    }
    *qr = -1; *ir = -1;
    return false;
}

/*******************************************************************************************
 *                               Reply Methods                                             *
 ******************************************************************************************/

void Replica::replyPrepare(int32_t LeaderId, PrepareReply * preply) {
    // TODO - DONE

    serializeBuff *msg = serialize_message(PREPARE_REPLY, (void *) preply);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[LeaderId], msg, PREPARE_REPLY);
    }else{
        //preply->Marshal(Peers[LeaderId]);
        send_message(Peers[LeaderId], msg, PREPARE_REPLY);
    }
    if(ret == -1){
        destroyConnection(Peers[LeaderId]);
        if (Debug) {
            fprintf(stderr, "Replica %d disconnected when reply Prepare to it.\n", LeaderId);
        }
        Alive[LeaderId] = false;
    }

    delete msg;

//    if (!preply->Marshal(Peers[LeaderId])) {
//        destroyConnection(Peers[LeaderId]);
//        if (Debug) {
//            fprintf(stderr, "Replica %d disconnected when reply Prepare to it.\n", LeaderId);
//        }
//        Alive[LeaderId] = false;
//    }
}

//Reply PreAccept to command Leader;

void Replica::replyPreAccept(int32_t LeaderId, PreAcceptReply * preacply) {
    // TODO - DONE

    serializeBuff *msg = serialize_message(PREACCEPT_REPLY, (void *) preacply);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[LeaderId], msg, PREACCEPT_REPLY);
    }else{
        //preacply->Marshal(Peers[LeaderId]);
        send_message(Peers[LeaderId], msg, PREACCEPT_REPLY);
    }
    if(ret == -1){
        destroyConnection(Peers[LeaderId]);
        if (Debug) {
            fprintf(stderr, "Cannot send preAcceptReply(instance = %d) to replica %d\n", preacply->Instance, LeaderId);
        }
        Alive[LeaderId] = false;
    }

    delete msg;

//    if (!preacply->Marshal(Peers[LeaderId])) {
//        destroyConnection(Peers[LeaderId]);
//        if (Debug) {
//            fprintf(stderr, "Cannot send preAcceptReply(instance = %d) to replica %d\n", preacply->Instance, LeaderId);
//        }
//        Alive[LeaderId] = false;
//    }
//   DLOG_LOG(DEBUG, "Replica %d reply PreAccept to command leader: %d, Instance: %d", Id, LeaderId, preacply->Instance);
}

//Reply Accept to Command Leader

void Replica::replyAccept(int32_t LeaderId, AcceptReply * acr) {
    // TODO - DONE
    serializeBuff *msg = serialize_message(ACCEPT_REPLY, (void *) acr);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[LeaderId], msg, ACCEPT_REPLY);
    }else{
        //acr->Marshal(Peers[LeaderId]);
        send_message(Peers[LeaderId], msg, ACCEPT_REPLY);
    }
    if(ret == -1){
        destroyConnection(Peers[LeaderId]);
        if (Debug) {
            fprintf(stderr, "Cannot send AcceptReply(instance = %d) to replica %d\n", acr->Instance, LeaderId);
        }
        Alive[LeaderId] = false;
    }
    delete msg;

//    if (!acr->Marshal(Peers[LeaderId])) {
//        destroyConnection(Peers[LeaderId]);
//        if (Debug) {
//            fprintf(stderr, "Cannot send AcceptReply(instance = %d) to replica %d\n", acr->Instance, LeaderId);
//        }
//        Alive[LeaderId] = false;
//    }
//    DLOG_LOG(DEBUG, "Replica %d reply Accept to command leader: %d, Instance: %d", Id, LeaderId, acr->Instance);
}

void Replica::replyTryPreAccept(int32_t LeaderId, TryPreAcceptReply * ntpap) {
    // TODO - DONE
    serializeBuff *msg = serialize_message(TRY_PREACCEPT, (void *)ntpap);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[LeaderId], msg, TRY_PREACCEPT);
    }else{
        //ntpap->Marshal(Peers[LeaderId]);
        send_message(Peers[LeaderId], msg, TRY_PREACCEPT);
    }
    if(ret == -1){
        Alive[LeaderId] = false;
    }
    delete msg;
//    if (!ntpap->Marshal(Peers[LeaderId]))
//        Alive[LeaderId] = false;
}

void Replica::sendBeacon(int32_t peerId) {
    // TODO - DONE
    // RDMA_Handler w = peerWriters[peerId];
    Beacon_msg * beacon = new Beacon_msg(Id, CPUTicks());
    if (beacon == nullptr) {
        fprintf(stderr, "Cannot alloc space for Beacon in sendBeacon...\n");
        return;
    }
    // beacon.Marshal(w);
    // w.Flush();
    // TEST

    serializeBuff *msg = serialize_message(BEACON, (void *) beacon);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[peerId], msg, BEACON);
    }else{
       // beacon->Marshal(Peers[peerId]);
        send_message(Peers[peerId], msg, BEACON);
    }
    if(ret == -1){
        Alive[peerId] = false;
    }
    delete msg;
//    if (!beacon->Marshal(Peers[peerId]))
//        Alive[peerId] = false;
    delete beacon;
    beacon = nullptr;
}

void Replica::replyBeacon(Beacon_msg * beacon) {
    // TODO - DONE
    // RDMA_Handler w = peerWriters[beacon->Rid];
    Beacon_msg_reply * rb = new Beacon_msg_reply(beacon->timestamp);
    if (rb == nullptr) {
        fprintf(stderr, "Cannot alloc space for BeaconReply in replyBeacon...\n");
        return;
    }
    // rb.Marshal(w);
    // w.Flush();
    // TEST
    serializeBuff* msg = serialize_message(BEACON_REPLY, (void *) rb);
    int ret = 0;
    if(UsingRDMA){
        ret = send_message(rdma_handlers[beacon->Rid], msg, BEACON_REPLY);
    }else{
        //rb->Marshal(beacon->Rid);
        send_message(Peers[beacon->Rid], msg, BEACON_REPLY);
    }
    if(ret == -1){
        Alive[beacon->Rid] = false;
    }
    delete msg;
//    if (!rb->Marshal(Peers[beacon->Rid]))
//        Alive[beacon->Rid] = false;
    delete rb;
    rb = nullptr;
}

/*******************************************************************************************
*                                Helper non-class Functions                                *
*******************************************************************************************/

void updateDeferred(int32_t dr, int32_t di, int32_t r, int32_t i) {
    uint64_t daux = (((uint64_t)dr) << 32) | ((uint64_t)di);
    uint64_t  aux = (((uint64_t) r) << 32) | ((uint64_t) i);
    deferMap[aux] = daux;
}

bool deferredByInstance(int32_t q, int32_t i, int32_t * dq, int32_t *di) {
    uint64_t aux = (((uint64_t)q) << 32) | ((uint64_t)i);
    auto daux = deferMap.find(aux);
    if (daux == deferMap.end()) {
        *dq = 0;
        *di = 0;
        return false;
    }
    *dq = (int32_t)(daux->second >> 32);
    *di = (int32_t)(daux->second);
    return true;
}

uint64_t CPUTicks() {
    clock_t t = clock();
    return (uint64_t)t;
}

void LEPutUint32(char * buf, int32_t v) {
    buf[0] = (char)v;
    buf[1] = (char)(v >> 8);
    buf[2] = (char)(v >> 16);
    buf[3] = (char)(v >> 24);
}


int find_longest_queue(lf_queue *message_queues, int queue_num){    
    if(message_queues == nullptr){
        return -1;
    }    
    int ret =  -1;
    int max_len = 0;
    for(int i = 0; i < queue_num; i++){
        int len = lf_queue_len(*(message_queues + i));
        if(len > max_len){
            max_len = len;
            ret = i; 
        }
    }
//    ret = -1;
    return ret;
}




